testnet-laconicd-stack/cli/map-subscribers-to-participants.ts

import * as fs from 'fs';
import * as crypto from 'crypto';
import * as path from 'path';
import yargs from 'yargs';
import { hideBin } from 'yargs/helpers';
import { parse as csvParse } from 'csv-parse';
import * as csvWriter from 'csv-writer';
import dotenv from 'dotenv';

import { StargateClient } from '@cosmjs/stargate';
import { Registry } from '@cerc-io/registry-sdk';

dotenv.config();

const LACONICD_CHAIN_ID = process.env.LACONICD_CHAIN_ID || 'laconic_9000-1';

async function main(): Promise<void> {
  const argv = _getArgv();

  const registry = new Registry(LACONICD_GQL_ENDPOINT, LACONICD_RPC_ENDPOINT, LACONICD_CHAIN_ID);
  const client = await StargateClient.connect(LACONICD_RPC_ENDPOINT);

  const participants = await registry.getParticipants();
  const subscribers = await readSubscribers(argv.subscribersCsv);

  processSubscribers(client, participants, subscribers, argv.output);
}

async function readSubscribers(subscribersCsvPath: string): Promise<any> {
  const fileContent = fs.readFileSync(subscribersCsvPath, { encoding: 'utf-8' });
  const headers = ['subscriber_id', 'email', 'status', 'premium?', 'created_at', 'api_subscription_id'];

  return csvParse(fileContent, { delimiter: ',', columns: headers }).toArray();
}

function hashSubscriberId(subscriberId: string): string {
  return '0x' + crypto.createHash('sha256').update(subscriberId).digest('hex');
}

async function processSubscribers(client: StargateClient, participants: any[], subscribers: any[], outputPath: string) {
  // Map kyc_id to participant data
  const kycMap: Record<string, any> = {};
  participants.forEach((participant: any) => {
    kycMap[participant.kyc_id] = participant;
  });

  const onboardedSubscribers: any[] = [];
  subscribers.forEach(async (subscriber) => {
    const hashedSubscriberId = hashSubscriberId(subscriber['subscriber_id']);
    const participant = kycMap[hashedSubscriberId];
    if (!participant) {
      return;
    }

    const participantAddresss = participant['cosmos_address'];

    // Fetch participant's Laconic pubkey
    const participantAccount = await client.getAccount(participantAddresss);
    const participantPubkey = participantAccount?.pubkey;

    // Skip participant if pubkey not found
    // (account may have funds but hasn't done any tx till now)
    if (!participantPubkey) {
      return;
    }

    // Skip participant if an onboarding tx not found
    const onboardingTxs = await client.searchTx(`message.sender='${participantAddresss}' AND message.action='/cerc.onboarding.v1.MsgOnboardParticipant'`);
    if (onboardingTxs.length === 0) {
      return;
    }

    const latestOnboardingTx = onboardingTxs.reduce((prev, current) => {
      return current.height > prev.height ? current : prev;
    });

    const onboardedSubscriber = {
      subscriber_id: subscriber['subscriber_id'],
      email: subscriber['email'],
      status: subscriber['status'],
      'premium?': subscriber['premium?'],
      created_at: subscriber['created_at'],
      cosmos_address: participantAddresss,
      nitro_address: participant['nitro_address'],
      role: participant['role'],
      hashed_subscriber_id: participant['kyc_id'],
      laconic_pubkey: participantPubkey,
      onboarding_height: latestOnboardingTx.height
    };

    onboardedSubscribers.push(onboardedSubscriber);
  });

  const writer = csvWriter.createObjectCsvWriter({
    path: path.resolve(outputPath),
    header: [
      { id: 'subscriber_id', title: 'subscriber_id' },
      { id: 'email', title: 'email' },
      { id: 'status', title: 'status' },
      { id: 'premium?', title: 'premium?' },
      { id: 'created_at', title: 'created_at' },
      { id: 'cosmos_address', title: 'cosmos_address' },
      { id: 'nitro_address', title: 'nitro_address' },
      { id: 'role', title: 'role' },
      { id: 'hashed_subscriber_id', title: 'hashed_subscriber_id' },
      { id: 'laconic_pubkey', title: 'laconic_pubkey' },
      { id: 'onboarding_height', title: 'onboarding_height' },
    ],
    alwaysQuote: true
  });

  await writer.writeRecords(onboardedSubscribers);

  console.log(`Data has been written to ${path.resolve(outputPath)}`);
}

function _getArgv (): any {
  return yargs(hideBin(process.argv))
  .option('subscribersCsv', {
    alias: 's',
    type: 'string',
    demandOption: true,
    describe: 'Path to the subscribers CSV file',
  })
  .option('output', {
    alias: 'o',
    type: 'string',
    demandOption: true,
    describe: 'Path to the output CSV file',
  })
  .help()
  .argv;
}

main().catch(err => {
  console.log(err);
});