Sweep.ts

import pMap from 'p-map';
import pRetry from 'p-retry';
import { safePromise } from 'safep';
import { chain, values, chunk, flatten } from 'lodash';
import { fromEnv as configFromEnv, SlpDepositConfig } from './config';
import { slpTokens } from '../shared/slp-tokens';
import { runWorkerUntilShutdown } from '../shared/utils-node';
import { getBitcoinRpc } from '../shared/rpcs';
import {
  toBchAddress,
  fetchSlpBalancesAndUtxos,
  Utxo,
  getBitbox,
  fetchSlpBalancesAndUtxosForAddress,
  SlpUtxo,
} from '../shared/slp';
import { n } from '../shared/utils';

const debug = require('consol').debugger('sideshift:slp-deposit:sweep');

export const createSweep = ({ config }: { config: SlpDepositConfig }) => {
  const rpc = getBitcoinRpc('slp');

  const addPrivateKeyToUtxo = async <T extends Utxo>(utxo: T) =>
    ({
      ...utxo,
      wif: await rpc.dumpPrivateKey(utxo.cashAddress),
    } as T);

  const fetchAddressesWithUtxos = async () => {
    const utxos = await rpc.listUnspent();

    return chain(utxos)
      .map(_ => _.address)
      .uniq()
      .value();
  };

  const fetchSlpBalancesAndUtxosChunked = async (addresses: string[]) => {
    const CHUNK_SIZE = 10;
    const chunks = chunk(addresses, CHUNK_SIZE);

    debug(`Fetching utxos for ${addresses.length} addresses in ${chunks.length} chunks...`);

    const results = await pMap(chunks, addresses => fetchSlpBalancesAndUtxos({ addresses }), {
      concurrency: 5,
    });

    const result = flatten(results);

    debug(`Fetched ${results.length} utxos`);

    return result;
  };

  const fetchUtxoForFees = async () => {
    const satoshisRequired = 10 * 1000;

    const result = await fetchSlpBalancesAndUtxosForAddress({
      address: config.slpFundingAddress!,
    });

    const { nonSlpUtxos } = result;

    const utxo = chain(nonSlpUtxos)
      .filter(_ => _.satoshis >= satoshisRequired)
      .orderBy(_ => _.satoshis)
      .first()
      .value();

    return utxo ? await addPrivateKeyToUtxo(utxo) : undefined;
  };

  const tick = async () => {
    // Fetch all addresss that have unspent coins, excluding the SLP funding address
    const addresses = chain(await fetchAddressesWithUtxos())
      .without(toBchAddress(config.slpFundingAddress!))
      .value();

    // Fetch all utxos for those addressees
    const utxos = await fetchSlpBalancesAndUtxosChunked(addresses);

    for (const token of values(slpTokens)) {
      const { tokenId } = token;

      // TODO: Can be solved more elegant with groupBy
      const tokenUtxos: SlpUtxo[] = chain(utxos)
        .map(_ => _.result.slpTokenUtxos[tokenId])
        .flatten()
        .filter(_ => _ !== undefined)
        .value();

      const unitsToSend = tokenUtxos.reduce(
        (sum, units) => sum.plus(units.slpUtxoJudgementAmount),
        n(0)
      );

      if (unitsToSend.eq(0)) {
        continue;
      }

      const tokenUtxosWithWif = await pMap(tokenUtxos, addPrivateKeyToUtxo);

      const utxoForFees = await fetchUtxoForFees();

      if (utxoForFees === undefined) {
        console.error(`There are no unspent coins in ${config.slpFundingAddress} to use for fees`);
        continue;
      }

      const utxoForFeesWithWif = await addPrivateKeyToUtxo(utxoForFees);

      const amountToSend = unitsToSend.div(10 ** token.decimals);

      console.log(
        `Sweeping ${amountToSend.toString()} ${token.asset} from ${tokenUtxos.length} utxos...`
      );

      const sendOptions = [
        tokenId,
        [unitsToSend],
        [utxoForFeesWithWif, ...tokenUtxosWithWif],
        [config.slpFundingAddress],
        config.slpFundingAddress,
      ];

      const { bitboxNetwork } = getBitbox();

      const [error, txid] = await safePromise(bitboxNetwork.simpleTokenSend(...sendOptions));

      if (error) {
        // NOTE: There are a lot of errors to ignore, including network errors (502),
        // rate limiting, etc. It may be easier to allow some blind retry in the outer loop
        throw error;
      }

      console.log(`Sweep completed. Txid ${txid}`);
    }
  };

  const sweep = runWorkerUntilShutdown(() => pRetry(tick), config.sweepInterval);

  return sweep;
}