Use cursor for invoice subscription

prisma/migrations/20240104004135_invoice_confirmed_index/migration.sql

@@ -0,0 +1,5 @@
+-- AlterTable
+ALTER TABLE "Invoice" ADD COLUMN     "confirmedIndex" BIGINT;
+
+-- CreateIndex
+CREATE INDEX "Invoice.confirmedIndex_index" ON "Invoice"("confirmedIndex");

prisma/schema.prisma

@@ -538,6 +538,7 @@ model Invoice {
   bolt11         String
   expiresAt      DateTime
   confirmedAt    DateTime?
+  confirmedIndex BigInt?
   cancelled      Boolean   @default(false)
   msatsRequested BigInt
   msatsReceived  BigInt?
@@ -548,6 +549,7 @@ model Invoice {
 
   @@index([createdAt], map: "Invoice.created_at_index")
   @@index([userId], map: "Invoice.userId_index")
+  @@index([confirmedIndex], map: "Invoice.confirmedIndex_index")
 }
 
 model Withdrawl {

worker/index.js

@@ -91,7 +91,9 @@ async function work () {
 
   await boss.start()
 
-  subWrapper(subscribeToInvoices({ lnd }), 'invoice_updated', (inv) => checkInvoice({ data: { hash: inv.id, sub: true }, ...args }))
+  const [lastConfirmed] = await models.$queryRaw`SELECT "confirmedIndex" FROM "Invoice" ORDER BY "confirmedIndex" DESC NULLS LAST LIMIT 1`
+  subWrapper(subscribeToInvoices({ lnd, confirmed_after: lastConfirmed?.confirmedIndex }),
+    'invoice_updated', (inv) => checkInvoice({ data: { hash: inv.id, sub: true }, ...args }))
   await boss.work('checkInvoice', jobWrapper(checkInvoice))
 
   await boss.work('checkWithdrawal', jobWrapper(checkWithdrawal))

worker/wallet.js

@@ -30,7 +30,9 @@ export async function checkInvoice ({ data: { hash, isHeldSet, sub }, boss, mode
     // never mark hodl invoices as confirmed here because
     // we manually confirm them when we settle them
     await serialize(models,
-      models.$executeRaw`SELECT confirm_invoice(${inv.id}, ${Number(inv.received_mtokens)})`)
+      models.$executeRaw`SELECT confirm_invoice(${inv.id}, ${Number(inv.received_mtokens)}`,
+      models.invoice.update({ where: { hash }, data: { confirmedIndex: inv.confirmed_index } })
+    )
     if (sub) {
       // only send push notifications in the context of a LND subscription.
       // else, when this code is run from polling context, we would send another push notification.
@@ -61,7 +63,7 @@ export async function checkInvoice ({ data: { hash, isHeldSet, sub }, boss, mode
     // this is basically confirm_invoice without setting confirmed_at since it's not settled yet
     // and without setting the user balance since that's done inside the same tx as the HODL invoice action.
     await serialize(models,
-      models.invoice.update({ where: { hash }, data: { msatsReceived: Number(inv.received_mtokens), isHeld: true } }))
+      models.invoice.update({ where: { hash }, data: { msatsReceived: Number(inv.received_mtokens), isHeld: true, confirmedIndex: inv.confirmed_index } }))
     // remember that we already executed this if clause
     // (even though the query above is idempotent but imo, this makes the flow more clear)
     isHeldSet = true