stacker.news/worker/wallet.js

import serialize from '@/api/resolvers/serial.js'
import {
  getInvoice, getPayment, cancelHodlInvoice, deletePayment,
  subscribeToInvoices, subscribeToPayments, subscribeToInvoice
} from 'ln-service'
import { notifyDeposit, notifyWithdrawal } from '@/lib/webPush'
import { INVOICE_RETENTION_DAYS, LND_PATHFINDING_TIMEOUT_MS } from '@/lib/constants'
import { datePivot, sleep } from '@/lib/time.js'
import retry from 'async-retry'
import { addWalletLog } from '@/api/resolvers/wallet'
import { msatsToSats, numWithUnits } from '@/lib/format'
import {
  paidActionPaid, paidActionForwarded,
  paidActionFailedForward, paidActionHeld, paidActionFailed,
  paidActionForwarding,
  paidActionCanceling
} from './paidAction.js'

export async function subscribeToWallet (args) {
  await subscribeToDeposits(args)
  await subscribeToWithdrawals(args)
}

// lnd subscriptions can fail, so they need to be retried
function subscribeForever (subscribe) {
  retry(async bail => {
    let sub
    try {
      return await new Promise((resolve, reject) => {
        sub = subscribe(resolve, bail)
        if (!sub) {
          return bail(new Error('function passed to subscribeForever must return a subscription object or promise'))
        }
        if (sub.then) {
          // sub is promise
          sub.then(sub => sub.on('error', reject))
        } else {
          sub.on('error', reject)
        }
      })
    } catch (error) {
      console.error(error)
      throw new Error('error subscribing - trying again')
    } finally {
      sub?.removeAllListeners()
    }
  },
  // retry every .1-10 seconds forever
  { forever: true, minTimeout: 100, maxTimeout: 10000, onRetry: e => console.error(e.message) })
}

const logEvent = (name, args) => console.log(`event ${name} triggered with args`, args)
const logEventError = (name, error) => console.error(`error running ${name}`, error)

async function subscribeToDeposits (args) {
  const { models, lnd } = args

  subscribeForever(async () => {
    const [lastConfirmed] = await models.$queryRaw`
    SELECT "confirmedIndex"
    FROM "Invoice"
    ORDER BY "confirmedIndex" DESC NULLS LAST
    LIMIT 1`
    const sub = subscribeToInvoices({ lnd, confirmed_after: lastConfirmed?.confirmedIndex })

    sub.on('invoice_updated', async (inv) => {
      try {
        logEvent('invoice_updated', inv)
        if (inv.secret) {
          await checkInvoice({ data: { hash: inv.id }, ...args })
        } else {
          // this is a HODL invoice. We need to use SubscribeToInvoice which has is_held transitions
          // https://api.lightning.community/api/lnd/invoices/subscribe-single-invoice
          // SubscribeToInvoices is only for invoice creation and settlement transitions
          // https://api.lightning.community/api/lnd/lightning/subscribe-invoices
          subscribeToHodlInvoice({ hash: inv.id, ...args })
        }
      } catch (error) {
        logEventError('invoice_updated', error)
      }
    })

    return sub
  })

  // check pending deposits as a redundancy in case we failed to rehcord
  // an invoice_updated event
  await checkPendingDeposits(args)
}

function subscribeToHodlInvoice (args) {
  const { lnd, hash } = args

  subscribeForever((resolve, reject) => {
    const sub = subscribeToInvoice({ id: hash, lnd })

    sub.on('invoice_updated', async (inv) => {
      logEvent('hodl_invoice_updated', inv)
      try {
        await checkInvoice({ data: { hash: inv.id }, ...args })
        // after settle or confirm we can stop listening for updates
        if (inv.is_confirmed || inv.is_canceled) {
          resolve()
        }
      } catch (error) {
        logEventError('hodl_invoice_updated', error)
        reject(error)
      }
    })

    return sub
  })
}

export async function checkInvoice ({ data: { hash }, boss, models, lnd }) {
  const inv = await getInvoice({ id: hash, lnd })

  // invoice could be created by LND but wasn't inserted into the database yet
  // this is expected and the function will be called again with the updates
  const dbInv = await models.invoice.findUnique({
    where: { hash },
    include: {
      invoiceForward: {
        include: {
          withdrawl: true
        }
      }
    }
  })
  if (!dbInv) {
    console.log('invoice not found in database', hash)
    return
  }

  if (inv.is_confirmed) {
    if (dbInv.actionType) {
      return await paidActionPaid({ data: { invoiceId: dbInv.id }, models, lnd, boss })
    }

    // NOTE: confirm invoice prevents double confirmations (idempotent)
    // ALSO: is_confirmed and is_held are mutually exclusive
    // that is, a hold invoice will first be is_held but not is_confirmed
    // and once it's settled it will be is_confirmed but not is_held
    const [[{ confirm_invoice: code }]] = await serialize([
      models.$queryRaw`SELECT confirm_invoice(${inv.id}, ${Number(inv.received_mtokens)})`,
      models.invoice.update({ where: { hash }, data: { confirmedIndex: inv.confirmed_index } })
    ], { models })

    // don't send notifications for JIT invoices
    if (dbInv.preimage) return
    if (code === 0) {
      notifyDeposit(dbInv.userId, { comment: dbInv.comment, ...inv })
    }

    return await boss.send('nip57', { hash })
  }

  if (inv.is_held) {
    if (dbInv.actionType) {
      if (dbInv.invoiceForward) {
        if (dbInv.invoiceForward.withdrawl) {
          // transitions when held are dependent on the withdrawl status
          return await checkWithdrawal({ data: { hash: dbInv.invoiceForward.withdrawl.hash }, models, lnd, boss })
        }
        return await paidActionForwarding({ data: { invoiceId: dbInv.id }, models, lnd, boss })
      }
      return await paidActionHeld({ data: { invoiceId: dbInv.id }, models, lnd, boss })
    }
    // First query makes sure that after payment, JIT invoices are settled
    // within 60 seconds or they will be canceled to minimize risk of
    // force closures or wallets banning us.
    // Second query is basically confirm_invoice without setting confirmed_at
    // and without setting the user balance
    // those will be set when the invoice is settled by user action
    const expiresAt = new Date(Math.min(dbInv.expiresAt, datePivot(new Date(), { seconds: 60 })))
    return await serialize([
      models.$queryRaw`
      INSERT INTO pgboss.job (name, data, retrylimit, retrybackoff, startafter)
      VALUES ('finalizeHodlInvoice', jsonb_build_object('hash', ${hash}), 21, true, ${expiresAt})`,
      models.invoice.update({
        where: { hash },
        data: {
          msatsReceived: Number(inv.received_mtokens),
          expiresAt,
          isHeld: true
        }
      })
    ], { models })
  }

  if (inv.is_canceled) {
    if (dbInv.actionType) {
      return await paidActionFailed({ data: { invoiceId: dbInv.id }, models, lnd, boss })
    }

    return await serialize(
      models.invoice.update({
        where: {
          hash: inv.id
        },
        data: {
          cancelled: true
        }
      }), { models }
    )
  }
}

async function subscribeToWithdrawals (args) {
  const { lnd } = args

  // https://www.npmjs.com/package/ln-service#subscribetopayments
  subscribeForever(() => {
    const sub = subscribeToPayments({ lnd })

    sub.on('confirmed', async (payment) => {
      logEvent('confirmed', payment)
      try {
        await checkWithdrawal({ data: { hash: payment.id }, ...args })
      } catch (error) {
        logEventError('confirmed', error)
      }
    })

    sub.on('failed', async (payment) => {
      logEvent('failed', payment)
      try {
        await checkWithdrawal({ data: { hash: payment.id }, ...args })
      } catch (error) {
        logEventError('failed', error)
      }
    })

    return sub
  })

  // check pending withdrawals since they might have been paid while worker was down
  await checkPendingWithdrawals(args)
}

export async function checkWithdrawal ({ data: { hash }, boss, models, lnd }) {
  // get the withdrawl if pending or it's an invoiceForward
  const dbWdrwl = await models.withdrawl.findFirst({
    where: {
      hash,
      OR: [
        { status: null },
        { invoiceForward: { some: { } } }
      ]
    },
    include: {
      wallet: true,
      invoiceForward: {
        orderBy: { createdAt: 'desc' },
        include: {
          invoice: true
        }
      }
    }
  })

  // nothing to do if the withdrawl is already recorded and it isn't an invoiceForward
  if (!dbWdrwl) return

  let wdrwl
  let notSent = false
  try {
    wdrwl = await getPayment({ id: hash, lnd })
  } catch (err) {
    if (err[1] === 'SentPaymentNotFound' &&
      dbWdrwl.createdAt < datePivot(new Date(), { milliseconds: -LND_PATHFINDING_TIMEOUT_MS * 2 })) {
      // if the payment is older than 2x timeout, but not found in LND, we can assume it errored before lnd stored it
      notSent = true
    } else {
      throw err
    }
  }

  if (wdrwl?.is_confirmed) {
    if (dbWdrwl.invoiceForward.length > 0) {
      return await paidActionForwarded({ data: { invoiceId: dbWdrwl.invoiceForward[0].invoice.id }, models, lnd, boss })
    }

    const fee = Number(wdrwl.payment.fee_mtokens)
    const paid = Number(wdrwl.payment.mtokens) - fee
    const [{ confirm_withdrawl: code }] = await serialize(
      models.$queryRaw`SELECT confirm_withdrawl(${dbWdrwl.id}::INTEGER, ${paid}, ${fee})`,
      { models }
    )
    if (code === 0) {
      notifyWithdrawal(dbWdrwl.userId, wdrwl)
      if (dbWdrwl.wallet) {
        // this was an autowithdrawal
        const message = `autowithdrawal of ${
          numWithUnits(msatsToSats(paid), { abbreviate: false })} with ${
          numWithUnits(msatsToSats(fee), { abbreviate: false })} as fee`
        await addWalletLog({ wallet: dbWdrwl.wallet, level: 'SUCCESS', message }, { models })
      }
    }
  } else if (wdrwl?.is_failed || notSent) {
    if (dbWdrwl.invoiceForward.length > 0) {
      return await paidActionFailedForward({ data: { invoiceId: dbWdrwl.invoiceForward[0].invoice.id }, models, lnd, boss })
    }

    let status = 'UNKNOWN_FAILURE'; let message = 'unknown failure'
    if (wdrwl?.failed.is_insufficient_balance) {
      status = 'INSUFFICIENT_BALANCE'
      message = "you didn't have enough sats"
    } else if (wdrwl?.failed.is_invalid_payment) {
      status = 'INVALID_PAYMENT'
      message = 'invalid payment'
    } else if (wdrwl?.failed.is_pathfinding_timeout) {
      status = 'PATHFINDING_TIMEOUT'
      message = 'no route found'
    } else if (wdrwl?.failed.is_route_not_found) {
      status = 'ROUTE_NOT_FOUND'
      message = 'no route found'
    }

    const [{ reverse_withdrawl: code }] = await serialize(
      models.$queryRaw`
        SELECT reverse_withdrawl(${dbWdrwl.id}::INTEGER, ${status}::"WithdrawlStatus")`,
      { models }
    )

    if (code === 0 && dbWdrwl.wallet) {
      // add error into log for autowithdrawal
      await addWalletLog({
        wallet: dbWdrwl.wallet,
        level: 'ERROR',
        message: 'autowithdrawal failed: ' + message
      }, { models })
    }
  }
}

export async function autoDropBolt11s ({ models, lnd }) {
  const retention = `${INVOICE_RETENTION_DAYS} days`

  // This query will update the withdrawls and return what the hash and bol11 values were before the update
  const invoices = await models.$queryRaw`
    WITH to_be_updated AS (
      SELECT id, hash, bolt11
      FROM "Withdrawl"
      WHERE "userId" IN (SELECT id FROM users WHERE "autoDropBolt11s")
      AND now() > created_at + interval '${retention}'
      AND hash IS NOT NULL
      AND status IS NOT NULL
    ), updated_rows AS (
      UPDATE "Withdrawl"
      SET hash = NULL, bolt11 = NULL
      FROM to_be_updated
      WHERE "Withdrawl".id = to_be_updated.id)
    SELECT * FROM to_be_updated;`

  if (invoices.length > 0) {
    for (const invoice of invoices) {
      try {
        await deletePayment({ id: invoice.hash, lnd })
      } catch (error) {
        console.error(`Error removing invoice with hash ${invoice.hash}:`, error)
        await models.withdrawl.update({
          where: { id: invoice.id },
          data: { hash: invoice.hash, bolt11: invoice.bolt11 }
        })
      }
    }
  }
}

// The callback subscriptions above will NOT get called for JIT invoices that are already paid.
// So we manually cancel the HODL invoice here if it wasn't settled by user action
export async function finalizeHodlInvoice ({ data: { hash }, models, lnd, boss, ...args }) {
  const inv = await getInvoice({ id: hash, lnd })
  if (inv.is_confirmed) {
    return
  }

  const dbInv = await models.invoice.findUnique({
    where: { hash },
    include: {
      invoiceForward: {
        include: {
          withdrawl: true
        }
      }
    }
  })
  if (!dbInv) {
    console.log('invoice not found in database', hash)
    return
  }

  // if this is an actionType we need to cancel conditionally
  if (dbInv.actionType) {
    await paidActionCanceling({ data: { invoiceId: dbInv.id }, models, lnd, boss })
    await checkInvoice({ data: { hash }, models, lnd, ...args })
    return
  }

  await cancelHodlInvoice({ id: hash, lnd })
  // sync LND invoice status with invoice status in database
  await checkInvoice({ data: { hash }, models, lnd, ...args })
}

export async function checkPendingDeposits (args) {
  const { models } = args
  const pendingDeposits = await models.invoice.findMany({ where: { confirmedAt: null, cancelled: false } })
  for (const d of pendingDeposits) {
    try {
      await checkInvoice({ data: { hash: d.hash }, ...args })
      await sleep(10)
    } catch {
      console.error('error checking invoice', d.hash)
    }
  }
}

export async function checkPendingWithdrawals (args) {
  const { models } = args
  const pendingWithdrawals = await models.withdrawl.findMany({ where: { status: null } })
  for (const w of pendingWithdrawals) {
    try {
      await checkWithdrawal({ data: { hash: w.hash }, ...args })
      await sleep(10)
    } catch (err) {
      console.error('error checking withdrawal', w.hash, err)
    }
  }
}