Add support for transaction groups (#324)

* Catch failed transactions in worker

* Add transaction groups

src/db/transactions/getTxsByGroupId.ts

@@ -0,0 +1,26 @@
+import { PrismaTransaction } from "../../schema/prisma";
+import { getPrismaWithPostgresTx } from "../client";
+import { cleanTxs } from "./cleanTxs";
+
+interface GetTxsByGroupIdParams {
+  pgtx?: PrismaTransaction;
+  groupId: string;
+}
+
+export const getTxsByGroupId = async ({
+  pgtx,
+  groupId,
+}: GetTxsByGroupIdParams) => {
+  const prisma = getPrismaWithPostgresTx(pgtx);
+  const txs = await prisma.transactions.findMany({
+    where: {
+      groupId,
+    },
+  });
+
+  if (!txs) {
+    return [];
+  }
+
+  return cleanTxs(txs);
+};

src/prisma/migrations/20231203024522_groups/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "transactions" ADD COLUMN     "groupId" TEXT;

src/prisma/schema.prisma

@@ -96,6 +96,7 @@ model WalletNonce {
 
 model Transactions {
   id                        String    @id @default(uuid()) @map("id")
+  groupId                   String?   @map("groupId")
   chainId                   String    @map("chainId")
   // Shared
   data                      String?   @map("data")
@@ -110,7 +111,7 @@ model Transactions {
   gasPrice                  String?   @map("gasPrice")
   transactionType           Int?      @map("transactionType")
   transactionHash           String?   @map("transactionHash")
-  onChainTxStatus             Int?      @map("onChainTxStatus")
+  onChainTxStatus           Int?      @map("onChainTxStatus")
   // User Operation
   signerAddress             String?   @map("signerAddress")
   accountAddress            String?   @map("accountAddress")

src/server/routes/backend-wallet/sendTransactionBatch.ts

@@ -29,6 +29,7 @@ const BodySchema = Type.Array(
 
 const ReplySchema = Type.Object({
   result: Type.Object({
+    groupId: Type.String(),
     queueIds: Type.Array(Type.String()),
   }),
 });
@@ -61,7 +62,9 @@ export async function sendTransactionBatch(fastify: FastifyInstance) {
       const fromAddress = req.headers["x-backend-wallet-address"] as string;
       const chainId = await getChainIdFromChain(chain);
 
+      const groupId = uuidv4();
       const data = txs.map((tx) => ({
+        groupId,
         id: uuidv4(),
         chainId: chainId.toString(),
         fromAddress,
@@ -75,6 +78,7 @@ export async function sendTransactionBatch(fastify: FastifyInstance) {
 
       res.status(StatusCodes.OK).send({
         result: {
+          groupId,
           queueIds: data.map((tx) => tx.id.toString()),
         },
       });

src/server/routes/index.ts

@@ -98,6 +98,7 @@ import { sendTransactionBatch } from "./backend-wallet/sendTransactionBatch";
 import { healthCheck } from "./health";
 import { home } from "./home";
 import { updateRelayer } from "./relayer/update";
+import { checkGroupStatus } from "./transaction/group";
 
 export const withRoutes = async (fastify: FastifyInstance) => {
   // Backend Wallets
@@ -186,6 +187,7 @@ export const withRoutes = async (fastify: FastifyInstance) => {
   await fastify.register(checkTxStatus);
   await fastify.register(getAllTx);
   await fastify.register(getAllDeployedContracts);
+  await fastify.register(checkGroupStatus);
   await fastify.register(retryTransaction);
   await fastify.register(cancelTransaction);
 

src/server/routes/transaction/group.ts

@@ -0,0 +1,43 @@
+import { Static, Type } from "@sinclair/typebox";
+import { FastifyInstance } from "fastify";
+import { StatusCodes } from "http-status-codes";
+import { getTxsByGroupId } from "../../../db/transactions/getTxsByGroupId";
+import { standardResponseSchema } from "../../schemas/sharedApiSchemas";
+import { transactionResponseSchema } from "../../schemas/transaction";
+
+const ParamsSchema = Type.Object({
+  groupId: Type.String(),
+});
+
+const ReplySchema = Type.Object({
+  result: Type.Array(transactionResponseSchema),
+});
+
+export async function checkGroupStatus(fastify: FastifyInstance) {
+  fastify.route<{
+    Params: Static<typeof ParamsSchema>;
+    Reply: Static<typeof ReplySchema>;
+  }>({
+    method: "GET",
+    url: "/transaction/status/group/:groupId",
+    schema: {
+      summary: "Get transaction status for a group",
+      description: "Get the status for a transaction group.",
+      tags: ["Transaction"],
+      operationId: "status",
+      params: ParamsSchema,
+      response: {
+        ...standardResponseSchema,
+        [StatusCodes.OK]: ReplySchema,
+      },
+    },
+    handler: async (req, res) => {
+      const { groupId } = req.params;
+      const txs = await getTxsByGroupId({ groupId });
+
+      res.status(StatusCodes.OK).send({
+        result: txs,
+      });
+    },
+  });
+}

src/worker/tasks/processTx.ts

@@ -230,19 +230,22 @@ export const processTx = async () => {
           }
 
           // Send all the transactions as one batch request
-          const res = await fetch(provider.connection.url, {
-            method: "POST",
-            headers: {
-              "Content-Type": "application/json",
-              ...(provider.connection.url.includes("rpc.thirdweb.com")
-                ? {
-                    "x-secret-key": env.THIRDWEB_API_SECRET_KEY,
-                  }
-                : {}),
-            },
-            body: JSON.stringify(rpcRequests),
-          });
-          const rpcResponses: RpcResponse[] = await res.json();
+          let rpcResponses: RpcResponse[] = [];
+          if (rpcRequests.length > 0) {
+            const res = await fetch(provider.connection.url, {
+              method: "POST",
+              headers: {
+                "Content-Type": "application/json",
+                ...(provider.connection.url.includes("rpc.thirdweb.com")
+                  ? {
+                      "x-secret-key": env.THIRDWEB_API_SECRET_KEY,
+                    }
+                  : {}),
+              },
+              body: JSON.stringify(rpcRequests),
+            });
+            rpcResponses = await res.json();
+          }
 
           // Check how many transactions succeeded and increment nonce
           const incrementNonce = rpcResponses.reduce((acc, curr) => {