Add flush mechanism

packages/randomness-service/src/index.ts

@@ -23,6 +23,7 @@ class RandomnessService {
             chain: anvil,
             abis: abis,
             gasEstimator: new CustomGasEstimator(),
+            rpcAllowDebug: true,
         })
         this.commitmentTransactionFactory = new CommitmentTransactionFactory(
             this.txm,

packages/transaction-manager/lib/Transaction.ts

@@ -99,6 +99,14 @@ export class Transaction {
 
     readonly attempts: Attempt[]
 
+    // Whether the transaction has been updated and needs to be flushed to the database.
+    // This field is not persisted in the database.
+    pendingFlush: boolean
+
+    // This is true if the transaction has never been persisted to the database yet.
+    // This field is not persisted in the database.
+    notPersisted: boolean
+
     createdAt: Date
 
     updatedAt: Date
@@ -110,27 +118,31 @@ export class Transaction {
     readonly metadata: Record<string, unknown>
 
     constructor({
-        intentId,
-        from,
-        chainId,
         address,
         functionName,
         contractName,
         args,
         deadline,
+        metadata,
+        intentId,
+        from,
+        chainId,
         status,
         attempts,
         createdAt,
         updatedAt,
-        metadata,
+        pendingFlush,
+        notPersisted,
     }: TransactionConstructorConfig & {
+        intentId?: UUID
         from: Address
         chainId: number
-        intentId?: UUID
         status?: TransactionStatus
         attempts?: Attempt[]
         createdAt?: Date
         updatedAt?: Date
+        pendingFlush?: boolean
+        notPersisted?: boolean
     }) {
         this.intentId = intentId ?? createUUID()
         this.from = from
@@ -145,19 +157,21 @@ export class Transaction {
         this.createdAt = createdAt ?? new Date()
         this.updatedAt = updatedAt ?? new Date()
         this.metadata = metadata ?? {}
+        this.pendingFlush = pendingFlush === undefined ? true : pendingFlush
+        this.notPersisted = notPersisted === undefined ? true : notPersisted
     }
 
     addAttempt(attempt: Attempt): void {
         this.attempts.push(attempt)
-        this.updatedAt = new Date()
+        this.markUpdated()
     }
 
     removeAttempt(hash: Hash): void {
         const index = this.attempts.findIndex((attempt) => attempt.hash === hash)
         if (index > -1) {
             this.attempts.splice(index, 1)
         }
-        this.updatedAt = new Date()
+        this.markUpdated()
     }
 
     getInAirAttempts(): Attempt[] {
@@ -172,7 +186,7 @@ export class Transaction {
 
     changeStatus(status: TransactionStatus): void {
         this.status = status
-        this.updatedAt = new Date()
+        this.markUpdated()
         eventBus.emit(Topics.TransactionStatusChanged, {
             transaction: this,
         })
@@ -186,6 +200,22 @@ export class Transaction {
         return this.attempts[this.attempts.length - 1]
     }
 
+    markFlushed(): void {
+        this.pendingFlush = false
+
+        if (this.notPersisted) {
+            this.notPersisted = false
+        }
+    }
+
+    private markUpdated(): void {
+        this.updatedAt = new Date()
+
+        if (this.pendingFlush === false) {
+            this.pendingFlush = true
+        }
+    }
+
     toDbRow(): Insertable<TransactionTable> {
         return {
             intentId: this.intentId,
@@ -212,6 +242,8 @@ export class Transaction {
             metadata: row.metadata ? JSON.parse(row.metadata, bigIntReviver) : undefined,
             createdAt: new Date(row.createdAt),
             updatedAt: new Date(row.updatedAt),
+            notPersisted: false,
+            pendingFlush: false,
         })
     }
 }

packages/transaction-manager/lib/TransactionRepository.ts

@@ -59,59 +59,35 @@ export class TransactionRepository {
     }
 
     async saveTransactions(transactions: Transaction[]): Promise<Result<void, Error>> {
-        const result = await ResultAsync.fromPromise(
-            db
-                .insertInto("transaction")
-                .values(transactions.map((t) => t.toDbRow()))
-                .execute(),
-            unknownToError,
-        )
+        const transactionsToFlush = transactions.filter((t) => t.pendingFlush)
 
-        if (result.isOk()) {
-            for (const transaction of transactions) {
-                if (NotFinalizedStatuses.includes(transaction.status)) {
-                    this.notFinalizedTransactions.push(transaction)
-                }
-            }
-        }
-        return result.map(() => undefined)
-    }
+        const notPersistedTransactions = transactions.filter((t) => t.notPersisted)
 
-    async updateTransaction(transaction: Transaction): Promise<Result<undefined, Error>> {
-        const result = await ResultAsync.fromPromise(
-            db
-                .updateTable("transaction")
-                .set(transaction.toDbRow())
-                .where("intentId", "=", transaction.intentId)
-                .execute(),
-            unknownToError,
-        )
-
-        this.notFinalizedTransactions = this.notFinalizedTransactions.filter((transaction) =>
-            NotFinalizedStatuses.includes(transaction.status),
-        )
-
-        return result.map(() => undefined)
-    }
-
-    async updateTransactions(transactions: Transaction[]): Promise<Result<void, Error>> {
         const result = await ResultAsync.fromPromise(
             db.transaction().execute(async (dbTransaction) => {
-                const promises = transactions.map((t) =>
-                    dbTransaction
-                        .updateTable("transaction")
-                        .set(t.toDbRow())
-                        .where("intentId", "=", t.intentId)
-                        .execute(),
-                )
+                const promises = transactionsToFlush.map((t) => {
+                    if (t.notPersisted) {
+                        dbTransaction.insertInto("transaction").values(t.toDbRow()).execute()
+                    } else {
+                        dbTransaction
+                            .updateTable("transaction")
+                            .set(t.toDbRow())
+                            .where("intentId", "=", t.intentId)
+                            .execute()
+                    }
+                })
                 await Promise.all(promises)
             }),
             unknownToError,
         )
 
-        this.notFinalizedTransactions = this.notFinalizedTransactions.filter((transaction) =>
-            NotFinalizedStatuses.includes(transaction.status),
-        )
+        if (result.isOk()) {
+            this.notFinalizedTransactions.push(...notPersistedTransactions)
+            this.notFinalizedTransactions = this.notFinalizedTransactions.filter((transaction) =>
+                NotFinalizedStatuses.includes(transaction.status),
+            )
+            transactions.forEach((t) => t.markFlushed())
+        }
 
         return result
     }

packages/transaction-manager/lib/TransactionSubmitter.ts

@@ -119,7 +119,7 @@ export class TransactionSubmitter {
             gas: transactionRequest.gas,
         })
 
-        const updateResult = await this.txmgr.transactionRepository.updateTransaction(transaction)
+        const updateResult = await this.txmgr.transactionRepository.saveTransactions([transaction])
 
         if (updateResult.isErr()) {
             transaction.removeAttempt(hash)

packages/transaction-manager/lib/TxMonitor.ts

@@ -150,7 +150,7 @@ export class TxMonitor {
         await Promise.all(promises)
 
         const result = await ResultAsync.fromPromise(
-            this.transactionManager.transactionRepository.updateTransactions(transactions),
+            this.transactionManager.transactionRepository.saveTransactions(transactions),
             unknownToError,
         )