LN-Channel-Opener.js

'use strict'
const Db = require('./src/DB/DB')
const Order = require('./src/Orders/Order')
const config = require('./config/server.json')
const async = require('async')
const { Client: GrenacheClient } = require('blocktank-worker')
const { lnWorker, callWorker } = require('./src/util/common-workers')
const { find } = require('lodash')
const {parseChannelOpenErr, chanErrors: errors} = require("./src/util/channel-opening-errors")

console.log('Starting Channel Opener...')

const MAX_CHANNEL_OPEN_ATTEMPT = config.constants.max_attempt_channel_open
const MAX_CHAN_FEE = 200

Db((err) => {
  if (err) throw err
  console.log('Started database')
})

async function getPaidOrders () {
  const db = await Db()
  return db.LnChannelOrders.find({
    state: Order.ORDER_STATES.URI_SET,
    created_at: { $gte: Date.now() - 172800000 }
  }).limit(10000000).toArray()
}

async function getProducts (productIds) {
  const db = await Db()
  return db.Inventory.find({ _id: { $in: productIds } }).toArray()
}

async function updateOrders (orders) {
  
  const logInfo = []
  
  const update = await Promise.all(orders.map(async ({ order, result }) => {
    result.ts = Date.now()
    let state
    if (order.order_result.length === MAX_CHANNEL_OPEN_ATTEMPT || result.giveup) {
      // GIVE UP OPENING CHANNEL
      // Tried too many times or node has issues
      state = Order.ORDER_STATES.GIVE_UP
      order.order_result.push(result)
      log.push([
        order._id,
        "gave up",
        result
      ])
    } else if (result.channel_tx && result.channel_tx.transaction_id) {
      // CHANNEL IS OPENING
      state = Order.ORDER_STATES.OPENING
      order.channel_open_tx = result.channel_tx
      alert('info', `Opening Channel: ${order._id} - txid: ${JSON.stringify(result.channel_tx, null, 2)}`)
    } else {
      if (order.order_result.length === 0 || order.order_result[order.order_result.length - 1].channel_error !== result.channel_error) {
        order.order_result.push(result)
      } else {
        order.order_result[order.order_result.length - 1] = result
      }
      log.push([
        order._id,
        "retry",
        result
      ])
      state = order.state
    }

    if (result.error) {
      console.log('Failing to open channel: ', order._id)
    }

    await Order.updateOrder(order._id, {
      state,
      order_result: order.order_result,
      channel_open_tx: order.channel_open_tx
    })
  }))

  //alert("info",`Channel opening attempt result:\n${JSON.stringify(logInfo)}`)
  return update 
}

function parseChannelOptions (product, order) {
  if (product.product_type === 'LN_CHANNEL') {
    if (order.remote_balance > order.local_balance) {
      return false
    }

    if(order.remote_balance >= order.local_balance){
      return false
    }

    return {
      remote_amt: order.remote_balance,
      local_amt: order.local_balance
    }
  }
  return false
}

function addPeer ({ order, product }, cb) {
  lnWorker('addPeer', {
    socket: order.remote_node.addr,
    public_key: order.remote_node.public_key
  }, (err, data) => {
    if (err) {
      console.log('Adding peer failed. Could already be connected')
      console.log(err)
    }
    cb(null, data)
  })
}

function channelOpener () {
  let count = 0
  setInterval(() => {
    count = 0
  }, 60000)
  const max = 10

  async function openChannel ({ order, product }, cb) {
    if (count >= max) {
      alert('info', 'Channel opening is being throttled.')
      return cb(new Error('Throttled channel opening'))
    }
    count++
    const res = { order }
    const op = parseChannelOptions(product, order)
    if (!op) return cb(new Error('invalid order options'))

    if(op.remote_amt){
      op.give_tokens = op.remote_amt
      if(op.give_tokens < 0) return cb(new Error('Invalid channel balance amounts'))
      op.local_amt = op.local_amt + op.remote_amt
    }

    const chanOpenConfig = {
      local_amt:op.local_amt,
      remote_amt: op.remote_amt,
      remote_pub_key: order.remote_node.public_key,
      is_private: order.private_channel
    }

    try {
      chanOpenConfig.fee_rate = await getOpeningFee()
    } catch(err){
      console.log("Failed to get chan opening fee")
      return cb(err)
    }
    
    if(chanOpenConfig.fee_rate >= MAX_CHAN_FEE) return cb(chanErrors.FEE_TOO_HIGH(["FEE IS HIGHER THAN "+MAX_CHAN_FEE]))


    console.log(`Opening LN Channel to: ${JSON.stringify(chanOpenConfig,null,2)}`)

    lnWorker('openChannel', chanOpenConfig, (err, data) => {
      if (err) {
        const chanErr = parseChannelOpenErr(err, {
          remote_node : order.remote_node
        })
        res.result = { error: chanErr.toString() }
        console.log('Failed to open channel', order._id, chanErr.toString())
        return cb(null, res)
      }
      if (!data.transaction_id) {
        console.log('Failed to open channel, no txid:', order._id)
        res.result = { error : chanErrors.NO_TX_ID([err,data]) }
        return cb(null, res)
      }
      data.fee_rate = chanOpenConfig.fee_rate
      res.result = { channel_tx: data }
      cb(null, res)
    })
  }

  return openChannel
}


async function getOpeningFee(cb) {
  return new Promise((accept, reject) =>{
    callWorker('svc:btc:mempool', 'getCurrrentFeeThreshold', [{}], (err,data)=>{
      if(err) return reject(err)
      accept(data.min_fee)
    })
  })
}


function alert (level, msg) {
  gClient.send('svc:monitor:slack', [level, 'ln_channel', msg], () => {})
}

const gClient = new GrenacheClient()
async function main (cb) {
  const orders = await getPaidOrders()
  if (orders.length === 0) {
    console.log(`No orders to process. ${Date.now()}`)
    return cb()
  }
  const openChannel = channelOpener()

  const products = await getProducts(orders.map(({ product_id }) => product_id))

  console.log(`Processing ${orders.length} for ${products.length} Products`)

  async.mapSeries(orders, (order, next) => {
    const product = find(products, ({ _id }) => order.product_id.equals(_id))
    if (!product) return next(new Error('Failed to find product'))
    addPeer({ product, order }, () => {
      openChannel({ product, order }, next)
    })
  }, async (err, data) => {
    if (err) {
      console.log('Error processing orders', err)
      return cb(err)
    }
    try {
      await updateOrders(data)
    } catch (err) {
      console.log('Failed to update orders', err)
    }
    cb()
  })
}

let running = false
setInterval(() => {
  if (running) {
    console.log('Channel opener is already running.')
    return
  }
  running = true
  try {
    main(() => {
      running = false
    })
  } catch (err) {
    console.log('Channel opener failed')
    console.log(err)
    running = false
  }
}, 5000)