Merge pull request #202 from pheuberger/cronjob-safe-queue

Run cron job to start Safe signature requests queue processing

package.json

@@ -62,6 +62,7 @@
     "kysely": "^0.27.4",
     "lodash": "^4.17.21",
     "lru-cache": "^11.0.0",
+    "node-cron": "^3.0.3",
     "pg": "^8.12.0",
     "reflect-metadata": "^0.2.2",
     "rollup": "^4.12.0",
@@ -93,6 +94,7 @@
     "@swc/cli": "^0.3.12",
     "@swc/core": "^1.4.15",
     "@types/body-parser": "^1.19.5",
+    "@types/node-cron": "^3.0.11",
     "@types/pg": "^8.11.6",
     "@types/sinon": "^17.0.2",
     "@types/swagger-ui-express": "^4.1.6",

src/controllers/SignatureRequestController.ts

@@ -11,7 +11,7 @@ import {
 
 import { SupabaseDataService } from "../services/SupabaseDataService.js";
 import { verifyAuthSignedData } from "../utils/verifyAuthSignedData.js";
-import { SignatureRequestProcessor } from "../services/SignatureRequestProcessor.js";
+import SignatureRequestProcessor from "../services/SignatureRequestProcessor.js";
 
 interface CancelSignatureRequest {
   signature: string;

src/cron/SignatureRequestProcessing.ts

@@ -0,0 +1,31 @@
+import cron from "node-cron";
+
+import SignatureRequestProcessor from "../services/SignatureRequestProcessor.js";
+
+export default class SignatureRequestProcessorCron {
+  private static instance: SignatureRequestProcessorCron;
+  private processor: SignatureRequestProcessor;
+
+  private constructor() {
+    this.processor = SignatureRequestProcessor.getInstance();
+    this.setupCronJob();
+  }
+
+  private setupCronJob() {
+    // Run every 30 seconds
+    cron.schedule("*/30 * * * * *", async () => {
+      try {
+        await this.processor.processPendingRequests();
+      } catch (error) {
+        console.error("Error in signature request processor cron job:", error);
+      }
+    });
+  }
+
+  public static start(): void {
+    if (!SignatureRequestProcessorCron.instance) {
+      SignatureRequestProcessorCron.instance =
+        new SignatureRequestProcessorCron();
+    }
+  }
+}

src/index.ts

@@ -1,63 +1,61 @@
-import './instrument.js';
-import express, {type Express} from "express";
+import "./instrument.js";
+import express, { type Express } from "express";
 import "reflect-metadata";
 import cors from "cors";
-import {assertExists} from "./utils/assertExists.js";
-import {yoga} from "./client/graphql.js";
+import { assertExists } from "./utils/assertExists.js";
+import { yoga } from "./client/graphql.js";
 import swaggerUi from "swagger-ui-express";
-import swaggerJson from "./__generated__/swagger.json" assert {type: "json"}
-import {RegisterRoutes} from "./__generated__/routes/routes.js";
-import * as Sentry from '@sentry/node';
+import swaggerJson from "./__generated__/swagger.json" assert { type: "json" };
+import { RegisterRoutes } from "./__generated__/routes/routes.js";
+import * as Sentry from "@sentry/node";
+import SignatureRequestProcessorCron from "./cron/SignatureRequestProcessing.js";
 
 // @ts-expect-error BigInt is not supported by JSON
 BigInt.prototype.toJSON = function () {
-    const int = Number.parseInt(this.toString());
-    return int ?? this.toString();
+  const int = Number.parseInt(this.toString());
+  return int ?? this.toString();
 };
 
 // @ts-expect-error BigInt is not supported by JSON
 BigInt.prototype.fromJSON = function () {
-    return BigInt(this.toString());
+  return BigInt(this.toString());
 };
 
 const PORT = assertExists(process.env.PORT, "PORT");
 
 const app: Express = express();
 
-app.use(express.urlencoded({extended: true, limit: '1mb'}));
-app.use(express.json({limit: '1mb'}));
+app.use(express.urlencoded({ extended: true, limit: "1mb" }));
+app.use(express.json({ limit: "1mb" }));
 app.use(cors());
 
-app.get('/health', (req, res) => {
-    const data = {
-        uptime: process.uptime(),
-        message: 'OK',
-        date: new Date()
-    }
+app.get("/health", (req, res) => {
+  const data = {
+    uptime: process.uptime(),
+    message: "OK",
+    date: new Date(),
+  };
 
-    res.status(200).send(data);
+  res.status(200).send(data);
 });
 
 // Bind GraphQL Yoga to the graphql endpoint to avoid rendering the playground on any path
 app.use(yoga.graphqlEndpoint, yoga);
 
-app.use(
-    "/spec",
-    swaggerUi.serve,
-    swaggerUi.setup(swaggerJson)
-);
+app.use("/spec", swaggerUi.serve, swaggerUi.setup(swaggerJson));
 
 RegisterRoutes(app);
 
 // The error handler must be registered before any other error middleware and after all controllers
 Sentry.setupExpressErrorHandler(app);
 
-app.listen(PORT, () => {
-    console.log(
-        `🕸️ Running a GraphQL API server at http://localhost:${PORT}/v1/graphql`
-    );
+// Start Safe signature request processing cron job
+SignatureRequestProcessorCron.start();
 
-    console.log(`🚀 Running Swagger docs at http://localhost:${PORT}/spec`);
+app.listen(PORT, () => {
+  console.log(
+    `🕸️ Running a GraphQL API server at http://localhost:${PORT}/v1/graphql`,
+  );
 
+  console.log(`🚀 Running Swagger docs at http://localhost:${PORT}/spec`);
 });
-

src/services/SignatureRequestProcessor.ts

@@ -8,7 +8,7 @@ import { SafeApiQueue } from "./SafeApiQueue.js";
 type SignatureRequest =
   Database["public"]["Tables"]["signature_requests"]["Row"];
 
-export class SignatureRequestProcessor {
+export default class SignatureRequestProcessor {
   private static instance: SignatureRequestProcessor;
 
   private readonly dataService: SupabaseDataService;
@@ -22,7 +22,7 @@ export class SignatureRequestProcessor {
   async processPendingRequests(): Promise<void> {
     const pendingRequests = await this.getPendingRequests();
 
-    console.log(`Found ${pendingRequests.length} pending requests`);
+    console.log(`Found ${pendingRequests.length} pending signature requests`);
 
     for (const request of pendingRequests) {
       const command = getCommand(request);