run noncritical side effects outside critical path of paid action (#1464)

* run noncritical side effects outside critical path of paid action

* fix item fetching of zap side effect

* fix vapid pubkey env var name in readme

README.md

@@ -429,7 +429,7 @@ GITHUB_SECRET=<Client secret>
 
 ## Enabling web push notifications
 
-To enable Web Push locally, you will need to set the `VAPID_*` env vars. `VAPID_MAILTO` needs to be an email address using the `mailto:` scheme. For `NEXT_PUBLIC_VAPID_KEY` and `VAPID_PRIVKEY`, you can run `npx web-push generate-vapid-keys`.
+To enable Web Push locally, you will need to set the `VAPID_*` env vars. `VAPID_MAILTO` needs to be an email address using the `mailto:` scheme. For `NEXT_PUBLIC_VAPID_PUBKEY` and `VAPID_PRIVKEY`, you can run `npx web-push generate-vapid-keys`.
 
 <br>
 

api/paidAction/README.md

@@ -154,7 +154,11 @@ All functions have the following signature: `function(args: Object, context: Obj
        - it can optionally store in the invoice with the `invoiceId` the `actionId` to be able to link the action with the invoice regardless of retries
 - `onPaid`: called when the action is paid
     - if the action does not support optimism, this function is optional
-    - this function should be used to mark the rows created in `perform` as `PAID` and perform any other side effects of the action (like notifications or denormalizations)
+    - this function should be used to mark the rows created in `perform` as `PAID` and perform critical side effects of the action (like denormalizations)
+- `nonCriticalSideEffects`: called after the action is paid to run any side effects whose failure does not affect the action's execution
+    - this function is always optional
+    - it's passed the result of the action (or the action's paid invoice) and the current context
+    - this is where things like push notifications should be handled
 - `onFail`: called when the action fails
     - if the action does not support optimism, this function is optional
     - this function should be used to mark the rows created in `perform` as `FAILED`

api/paidAction/boost.js

@@ -38,7 +38,7 @@ export async function retry ({ invoiceId, newInvoiceId }, { tx, cost }) {
   return { id, sats: msatsToSats(cost), act: 'BOOST', path }
 }
 
-export async function onPaid ({ invoice, actId }, { models, tx }) {
+export async function onPaid ({ invoice, actId }, { tx }) {
   let itemAct
   if (invoice) {
     await tx.itemAct.updateMany({

api/paidAction/index.js

@@ -99,7 +99,7 @@ async function performFeeCreditAction (actionType, args, context) {
   const { me, models, cost } = context
   const action = paidActions[actionType]
 
-  return await models.$transaction(async tx => {
+  const result = await models.$transaction(async tx => {
     context.tx = tx
 
     await tx.user.update({
@@ -121,6 +121,11 @@ async function performFeeCreditAction (actionType, args, context) {
       paymentMethod: 'FEE_CREDIT'
     }
   }, { isolationLevel: Prisma.TransactionIsolationLevel.ReadCommitted })
+
+  // run non critical side effects in the background
+  // after the transaction has been committed
+  action.nonCriticalSideEffects?.(result.result, context).catch(console.error)
+  return result
 }
 
 async function performOptimisticAction (actionType, args, context) {

api/paidAction/itemCreate.js

@@ -154,15 +154,13 @@ export async function retry ({ invoiceId, newInvoiceId }, { tx }) {
 }
 
 export async function onPaid ({ invoice, id }, context) {
-  const { models, tx } = context
+  const { tx } = context
   let item
 
   if (invoice) {
     item = await tx.item.findFirst({
       where: { invoiceId: invoice.id },
       include: {
-        mentions: true,
-        itemReferrers: { include: { refereeItem: true } },
         user: true
       }
     })
@@ -173,8 +171,6 @@ export async function onPaid ({ invoice, id }, context) {
     item = await tx.item.findUnique({
       where: { id },
       include: {
-        mentions: true,
-        itemReferrers: { include: { refereeItem: true } },
         user: true,
         itemUploads: { include: { upload: true } }
       }
@@ -224,17 +220,30 @@ export async function onPaid ({ invoice, id }, context) {
         SELECT comment.created_at, comment.updated_at, ancestors.id, ancestors."userId",
           comment.id, comment."userId", nlevel(comment.path) - nlevel(ancestors.path)
         FROM ancestors, comment`
+  }
+}
 
+export async function nonCriticalSideEffects ({ invoice, id }, { models }) {
+  const item = await models.item.findFirst({
+    where: invoice ? { invoiceId: invoice.id } : { id: parseInt(id) },
+    include: {
+      mentions: true,
+      itemReferrers: { include: { refereeItem: true } },
+      user: true
+    }
+  })
+
+  if (item.parentId) {
     notifyItemParents({ item, models }).catch(console.error)
   }
-
   for (const { userId } of item.mentions) {
     notifyMention({ models, item, userId }).catch(console.error)
   }
   for (const { refereeItem } of item.itemReferrers) {
     notifyItemMention({ models, referrerItem: item, refereeItem }).catch(console.error)
   }
-  notifyUserSubscribers({ models: tx, item }).catch(console.error)
+
+  notifyUserSubscribers({ models, item }).catch(console.error)
   notifyTerritorySubscribers({ models, item }).catch(console.error)
 }
 

api/paidAction/itemUpdate.js

@@ -24,7 +24,7 @@ export async function getCost ({ id, boost = 0, uploadIds, bio }, { me, models }
 
 export async function perform (args, context) {
   const { id, boost = 0, uploadIds = [], options: pollOptions = [], forwardUsers: itemForwards = [], ...data } = args
-  const { tx, me, models } = context
+  const { tx, me } = context
   const old = await tx.item.findUnique({
     where: { id: parseInt(id) },
     include: {
@@ -63,12 +63,8 @@ export async function perform (args, context) {
 
   // we put boost in the where clause because we don't want to update the boost
   // if it has changed concurrently
-  const item = await tx.item.update({
+  await tx.item.update({
     where: { id: parseInt(id), boost: old.boost },
-    include: {
-      mentions: true,
-      itemReferrers: { include: { refereeItem: true } }
-    },
     data: {
       ...data,
       boost: {
@@ -151,6 +147,21 @@ export async function perform (args, context) {
 
   await performBotBehavior(args, context)
 
+  // ltree is unsupported in Prisma, so we have to query it manually (FUCK!)
+  return (await tx.$queryRaw`
+    SELECT *, ltree2text(path) AS path, created_at AS "createdAt", updated_at AS "updatedAt"
+    FROM "Item" WHERE id = ${parseInt(id)}::INTEGER`
+  )[0]
+}
+
+export async function nonCriticalSideEffects ({ invoice, id }, { models }) {
+  const item = await models.item.findFirst({
+    where: invoice ? { invoiceId: invoice.id } : { id: parseInt(id) },
+    include: {
+      mentions: true,
+      itemReferrers: { include: { refereeItem: true } }
+    }
+  })
   // compare timestamps to only notify if mention or item referral was just created to avoid duplicates on edits
   for (const { userId, createdAt } of item.mentions) {
     if (item.updatedAt.getTime() !== createdAt.getTime()) continue
@@ -160,12 +171,6 @@ export async function perform (args, context) {
     if (item.updatedAt.getTime() !== createdAt.getTime()) continue
     notifyItemMention({ models, referrerItem: item, refereeItem }).catch(console.error)
   }
-
-  // ltree is unsupported in Prisma, so we have to query it manually (FUCK!)
-  return (await tx.$queryRaw`
-    SELECT *, ltree2text(path) AS path, created_at AS "createdAt", updated_at AS "updatedAt"
-    FROM "Item" WHERE id = ${parseInt(id)}::INTEGER`
-  )[0]
 }
 
 export async function describe ({ id, parentId }, context) {

api/paidAction/zap.js

@@ -64,7 +64,7 @@ export async function retry ({ invoiceId, newInvoiceId }, { tx, cost }) {
   return { id, sats: msatsToSats(cost), act: 'TIP', path }
 }
 
-export async function onPaid ({ invoice, actIds }, { models, tx }) {
+export async function onPaid ({ invoice, actIds }, { tx }) {
   let acts
   if (invoice) {
     await tx.itemAct.updateMany({
@@ -114,7 +114,7 @@ export async function onPaid ({ invoice, actIds }, { models, tx }) {
 
   // perform denomormalized aggregates: weighted votes, upvotes, msats, lastZapAt
   // NOTE: for the rows that might be updated by a concurrent zap, we use UPDATE for implicit locking
-  const [item] = await tx.$queryRaw`
+  await tx.$queryRaw`
     WITH zapper AS (
       SELECT trust FROM users WHERE id = ${itemAct.userId}::INTEGER
     ), zap AS (
@@ -163,8 +163,14 @@ export async function onPaid ({ invoice, actIds }, { models, tx }) {
       SET "commentMsats" = "Item"."commentMsats" + ${msats}::BIGINT
       FROM zapped
       WHERE "Item".path @> zapped.path AND "Item".id <> zapped.id`
+}
 
-  notifyZapped({ models, item }).catch(console.error)
+export async function nonCriticalSideEffects ({ invoice, actIds }, { models }) {
+  const itemAct = await models.itemAct.findFirst({
+    where: invoice ? { invoiceId: invoice.id } : { id: { in: actIds } },
+    include: { item: true }
+  })
+  notifyZapped({ models, item: itemAct.item }).catch(console.error)
 }
 
 export async function onFail ({ invoice }, { tx }) {

worker/paidAction.js

@@ -134,7 +134,7 @@ async function performPessimisticAction ({ lndInvoice, dbInvoice, tx, models, ln
 }
 
 export async function paidActionPaid ({ data: { invoiceId, ...args }, models, lnd, boss }) {
-  return await transitionInvoice('paidActionPaid', {
+  const transitionedInvoice = await transitionInvoice('paidActionPaid', {
     invoiceId,
     fromState: ['HELD', 'PENDING', 'FORWARDED'],
     toState: 'PAID',
@@ -156,6 +156,14 @@ export async function paidActionPaid ({ data: { invoiceId, ...args }, models, ln
     },
     ...args
   }, { models, lnd, boss })
+
+  if (transitionedInvoice) {
+    // run non critical side effects in the background
+    // after the transaction has been committed
+    paidActions[transitionedInvoice.actionType]
+      .nonCriticalSideEffects?.({ invoice: transitionedInvoice }, { models, lnd })
+      .catch(console.error)
+  }
 }
 
 // this performs forward creating the outgoing payment