cli: contract based airdrop using disperse.app

This is much faster as we can bundle hundreds of transfers in a single transaction.
Add support for concurrency using p-queue and NonceManager.

Author: abarmat

cli/cli.ts

@@ -10,6 +10,7 @@ import { protocolCommand } from './commands/protocol'
 import { contractsCommand } from './commands/contracts'
 import { transferTeamTokensCommand } from './commands/transferTeamTokens'
 import { simulationCommand } from './commands/simulations'
+import { airdropCommand } from './commands/airdrop'
 
 import { cliOpts } from './constants'
 
@@ -29,5 +30,6 @@ yargs
   .command(contractsCommand)
   .command(transferTeamTokensCommand)
   .command(simulationCommand)
+  .command(airdropCommand)
   .demandCommand(1, 'Choose a command from the above list')
   .help().argv

cli/commands/airdrop.ts

@@ -0,0 +1,299 @@
+import consola from 'consola'
+import inquirer from 'inquirer'
+import fs from 'fs'
+import PQueue from 'p-queue'
+import yargs, { Argv } from 'yargs'
+import { parseGRT, formatGRT } from '@graphprotocol/common-ts'
+
+import { utils, BigNumber, Contract } from 'ethers'
+import { NonceManager } from '@ethersproject/experimental'
+
+import { sendTransaction } from '../network'
+import { loadEnv, CLIArgs, CLIEnvironment } from '../env'
+
+const logger = consola.create({})
+
+const { getAddress } = utils
+
+const DISPERSE_CONTRACT_ADDRESS = {
+  1: '0xD152f549545093347A162Dce210e7293f1452150',
+  4: '0xD152f549545093347A162Dce210e7293f1452150',
+  1337: '0xD152f549545093347A162Dce210e7293f1452150',
+}
+
+const DISPERSE_CONTRACT_ABI = [
+  {
+    constant: false,
+    inputs: [
+      { name: 'token', type: 'address' },
+      { name: 'recipients', type: 'address[]' },
+      { name: 'values', type: 'uint256[]' },
+    ],
+    name: 'disperseTokenSimple',
+    outputs: [],
+    payable: false,
+    stateMutability: 'nonpayable',
+    type: 'function',
+  },
+  {
+    constant: false,
+    inputs: [
+      { name: 'token', type: 'address' },
+      { name: 'recipients', type: 'address[]' },
+      { name: 'values', type: 'uint256[]' },
+    ],
+    name: 'disperseToken',
+    outputs: [],
+    payable: false,
+    stateMutability: 'nonpayable',
+    type: 'function',
+  },
+  {
+    constant: false,
+    inputs: [
+      { name: 'recipients', type: 'address[]' },
+      { name: 'values', type: 'uint256[]' },
+    ],
+    name: 'disperseEther',
+    outputs: [],
+    payable: true,
+    stateMutability: 'payable',
+    type: 'function',
+  },
+]
+
+interface AirdropRecipient {
+  address: string
+  amount: BigNumber
+  txHash?: string
+}
+
+const sure = async (message = 'Are you sure?'): Promise<boolean> => {
+  // Warn about changing ownership
+  const res = await inquirer.prompt({
+    name: 'confirm',
+    type: 'confirm',
+    message,
+  })
+  if (!res.confirm) {
+    consola.success('Cancelled')
+    return false
+  }
+  return true
+}
+
+const getDisperseContract = (chainID: number, provider) => {
+  return new Contract(DISPERSE_CONTRACT_ADDRESS[chainID], DISPERSE_CONTRACT_ABI, provider)
+}
+
+const loadRecipients = (path: string): Array<AirdropRecipient> => {
+  const data = fs.readFileSync(path, 'utf8')
+  const lines = data.split('\n').map((e) => e.trim())
+
+  const results: Array<AirdropRecipient> = []
+  for (const line of lines) {
+    const [address, amount, txHash] = line.split(',').map((e) => e.trim())
+
+    // Skip any empty value
+    if (!address) continue
+
+    // Test for zero amount and fail
+    const weiAmount = parseGRT(amount)
+    if (weiAmount.eq(0)) {
+      logger.fatal(`Error loading address "${address}" - amount is zero`)
+      process.exit(0)
+    }
+
+    // Validate address format
+    try {
+      getAddress(address)
+    } catch (err) {
+      // Full stop on error
+      logger.fatal(`Error loading address "${address}" please review the input file`)
+      process.exit(1)
+    }
+    results.push({ address, amount: weiAmount, txHash })
+  }
+  return results
+}
+
+const sumTotalAmount = (recipients: Array<AirdropRecipient>): BigNumber => {
+  let total = BigNumber.from(0)
+  for (const recipient of recipients) {
+    total = total.add(recipient.amount)
+  }
+  return total
+}
+
+const saveResumeList = (path: string, txHash: string, recipients: Array<AirdropRecipient>) => {
+  for (const recipient of recipients) {
+    const line = [recipient.address, formatGRT(recipient.amount), txHash].join(',') + '\n'
+    fs.writeFileSync(path, line, {
+      flag: 'a+',
+    })
+  }
+}
+
+const loadResumeList = (path: string): Array<AirdropRecipient> => {
+  try {
+    return loadRecipients(path)
+  } catch (err) {
+    if (err.code === 'ENOENT') {
+      logger.warn('No existing resumefile, one will be created')
+      return []
+    } else {
+      throw err
+    }
+  }
+}
+
+const createBatches = (
+  items: Array<AirdropRecipient>,
+  batchSize = 10,
+): Array<Array<AirdropRecipient>> => {
+  const remainingItems = Object.assign([], items)
+  const batches = []
+  while (remainingItems.length > 0) {
+    const batchItems = remainingItems.splice(0, batchSize)
+    batches.push(batchItems)
+    if (batchItems.length < batchSize) {
+      break
+    }
+  }
+  return batches
+}
+
+export const airdrop = async (cli: CLIEnvironment, cliArgs: CLIArgs): Promise<void> => {
+  const graphToken = cli.contracts.GraphToken
+
+  // Load data
+  const resumeList = loadResumeList(cliArgs.resumefile).map((r) => r.address)
+  const recipients = loadRecipients(cliArgs.recipients).filter(
+    (r) => !resumeList.includes(r.address),
+  )
+  const totalAmount = sumTotalAmount(recipients)
+
+  // Summary
+  logger.log(`# Batch Size: ${cliArgs.batchSize}`)
+  logger.log(`# Concurrency: ${cliArgs.concurrency}`)
+  logger.log(`> Token: ${graphToken.address}`)
+  logger.log(`> Distributing: ${formatGRT(totalAmount)} tokens (${totalAmount} wei)`)
+  logger.log(`> Resumelist: ${resumeList.length} addresses`)
+  logger.log(`> Recipients: ${recipients.length} addresses\n`)
+
+  // Validity check
+  if (totalAmount.eq(0)) {
+    logger.fatal('Cannot proceed with a distribution of zero tokens')
+    process.exit(1)
+  }
+
+  // Load airdrop contract
+  const disperseContract = getDisperseContract(cli.chainId, cli.wallet.provider)
+  if (!disperseContract.address) {
+    logger.fatal('Disperse contract not found. Please review your network settings.')
+    process.exit(1)
+  }
+
+  // Confirmation
+  logger.log('Are you sure you want to proceed with the distribution?')
+  if (!(await sure())) {
+    process.exit(1)
+  }
+
+  // Approve
+  logger.info('## Token approval')
+  const allowance = (
+    await graphToken.functions['allowance'](cli.wallet.address, disperseContract.address)
+  )[0]
+  if (allowance.gte(totalAmount)) {
+    logger.log('Already have enough allowance, no need to approve more...')
+  } else {
+    logger.log(
+      `Approve disperse:${disperseContract.address} for ${formatGRT(
+        totalAmount,
+      )} tokens (${totalAmount} wei)`,
+    )
+    await sendTransaction(
+      cli.wallet,
+      graphToken,
+      'approve',
+      ...[disperseContract.address, totalAmount],
+    )
+  }
+
+  // Distribute
+  logger.info('## Distribution')
+  const queue = new PQueue({ concurrency: cliArgs.concurrency })
+  const recipientsBatches = createBatches(recipients, cliArgs.batchSize)
+  const nonceManager = new NonceManager(cli.wallet) // Use NonceManager to send concurrent txs
+
+  let batchNum = 0
+  let recipientsCount = 0
+  for (const batch of recipientsBatches) {
+    queue.add(async () => {
+      const addressList = batch.map((r) => r.address)
+      const amountList = batch.map((r) => r.amount)
+
+      recipientsCount += addressList.length
+      batchNum++
+      logger.info(`Sending batch #${batchNum} : ${recipientsCount}/${recipients.length}`)
+      for (const recipient of batch) {
+        logger.log(
+          `  > Transferring ${recipient.address} => ${formatGRT(recipient.amount)} (${
+            recipient.amount
+          } wei)`,
+        )
+      }
+      try {
+        const receipt = await sendTransaction(
+          nonceManager,
+          disperseContract,
+          'disperseToken',
+          ...[graphToken.address, addressList, amountList],
+        )
+        saveResumeList(cliArgs.resumefile, receipt.transactionHash, batch)
+      } catch (err) {
+        logger.error(`Failed to send #${batchNum}`, err)
+      }
+    })
+  }
+  await queue.onIdle()
+}
+
+export const airdropCommand = {
+  command: 'airdrop',
+  describe: 'Airdrop tokens',
+  builder: (yargs: Argv): yargs.Argv => {
+    return yargs
+      .option('recipients', {
+        description: 'Path to the file with information for the airdrop. CSV file address,amount',
+        type: 'string',
+        requiresArg: true,
+        demandOption: true,
+      })
+      .option('resumefile', {
+        description:
+          'Path to the file used for resuming. Stores results with CSV format: address,amount,txHash',
+        type: 'string',
+        requiresArg: true,
+        demandOption: true,
+      })
+      .option('batch-size', {
+        description: 'Number of addresses to send in a single transaction',
+        type: 'number',
+        requiresArg: true,
+        demandOption: true,
+        default: 100,
+      })
+      .option('concurrency', {
+        description: 'Number of simultaneous transfers',
+        type: 'number',
+        requiresArg: true,
+        demandOption: false,
+        default: 1,
+      })
+  },
+  handler: async (argv: CLIArgs): Promise<void> => {
+    return airdrop(await loadEnv(argv), argv)
+  },
+}

cli/mockData/team-airdrop.csv

@@ -0,0 +1,15 @@
+0x8Df7727e3B5aCC57Be09578dc54CDD53a9dA156C,100
+0x0E322262016E94EBE6b17e8396C00D04538b3aED,100
+0x04604EcE79E9E0eeE7B1C262223614e02047B74f,100
+0xA8B2B5c22E5c13E9F789284b067736D906A5AFa9,100
+0x85FAC3f0734Ac360712caED2C0b9782133ed37Da,100
+0xbEb1Faa6E7e39c7d9BdaB03a7a362fE9d73D7C61,100
+0x1b037167C4b0584Ca5Ef6534648C38F496757FA5,100
+0xeF38F892E4722152fD8eDb50cD84a96344FD47Ce,100
+0x93606b27cB5e4c780883eC4F6b7Bed5f6572d1dd,100
+0xD31bC1e2a214066Bb2258dac5f43Ce75e5542Ab9,100
+0x055BCF2c2BC965Ac8EEAe3f95922D65EE45d3366,100
+0xc02624Affa30299fA164e7176A37610835A923A7,100
+0x9e0a3BD68620abD81226047cA0e1e0c63A2BE36B,100
+0x2EF7cD7a130a4bbbD637312a3B4A8E5365296e69,100
+0x2ACa45C45580C17E60E80673E2a2595d11009552,100

cli/network.ts

@@ -73,11 +73,11 @@ export const sendTransaction = async (
   // Send transaction
   let tx: ContractTransaction
   try {
-    tx = await contract.functions[fn](...params)
+    tx = await contract.connect(sender).functions[fn](...params)
   } catch (e) {
     if (e.code == 'UNPREDICTABLE_GAS_LIMIT') {
       logger.warn(`Gas could not be estimated - trying defaultOverrides`)
-      tx = await contract.functions[fn](...params, defaultOverrides())
+      tx = await contract.connect(sender).functions[fn](...params, defaultOverrides())
     } else {
       throw e
     }
@@ -87,7 +87,8 @@ export const sendTransaction = async (
       `It appears the function does not exist on this contract, or you have the wrong contract address`,
     )
   }
-  logger.log(`> Sent transaction ${fn}: ${params}, txHash: ${tx.hash}`)
+  logger.log(`> Sent transaction ${fn}: [${params}] txHash: ${tx.hash}`)
+
   // Wait for transaction to be mined
   const receipt = await sender.provider.waitForTransaction(tx.hash)
   const networkName = (await sender.provider.getNetwork()).name