riki.bio

This system is designed to index and monitor token movements on the Solana blockchain. It fetches transaction history, parses token-related instructions, and stores the information in a Firebase database. The system provides real-time updates and historical data for a specific token.

Monitor:

The main class that initializes and coordinates the indexing process:

Initializes the Database, Parser, Fetcher and Listener

There are two transaction entry points:

API server using Elysia that handles webhook requests, ie: transactions on real time
Initializes the monitoring process for a specific token by reading its history


class MonitorToken {
    private parser: Parser;
    private fetcher: Fetcher;
    private listener: Listener;
    private db: Database;
  
    constructor() {
	      this.db = new Database();
	      
	      this.parser = new Parser(this.db);
	      this.fetcher = new Fetcher(this.db);
	  
	      Object.assign(this.parser, { fetcher: this.fetcher });
	      Object.assign(this.fetcher, { parser: this.parser });
	  
	      this.listener = new Listener(this.parser);
    }

	  async init(token: string): Promise<void> {
        const mintAccountInfo = await config.RPC.getAccountInfo(new PublicKey(token));
        await this.parser.mint(token, mintAccountInfo);
        await this.fetcher.tokenMovements(token);
    }

    getListenerHandler() {
        return this.listener.getHandler();
    }
}

const monitor = new MonitorToken();
new Elysia()
    .use(monitor.getListenerHandler())
    .listen({ hostname: config.HOST, port: config.PORT }, async ({ hostname, port }: ListenOptions) => {
        console.log(`Running at http://${hostname}:${port}`);
        await monitor.init(config.TOKEN);
    });

Fetcher:

Responsible for retrieving transaction data from the mint to send them to the parser where the parsed events are stored. mintFromHistory is responsible for getting the mint from source and destination account on the transfer event (this instruction does not have this info in its context, as we only want to process the transfers of an specific mint, we need to check if this account is from our mint, in case this account does not longer exists the unique way to validate this is to read its history)


export class Fetcher {
  private db: Database;
  private parser!: Parser;

  constructor(db: Database) {
    this.db = db;
  }

  public async tokenMovements(account: string): Promise<void> {
    const pubkey = new PublicKey(account);
    await this.transactions(pubkey, async (transactions) => {
      await Promise.all(transactions.map(transaction => 
        this.parser.tokenMovements(transaction)
      ));
    }, 10);
  }

  public async mintFromHistory(keys: string[]): Promise<string> {
    const accountInfos = await config.RPC.getMultipleAccountsInfo(keys.map(x => new PublicKey(x)));
    
    for (const [_, accountInfo] of accountInfos.entries()) {
        if (accountInfo) return AccountLayout.decode(accountInfo.data).mint.toBase58();
    }

    for (const account of keys) {
        const pubkey = new PublicKey(account);
        const mint = await this.transactions(pubkey, async (transactions) => {
            for (const transaction of transactions) {
                const mint = await this.parser.mintFromHistory(transaction, account);
                if (mint) return mint;
            }
            return undefined;
        }, 5);

        if (mint) return mint;
    }

    return '';
  }

  private async transactions(
    pubkey: PublicKey,
    batchProcessor: (transactions: ParsedTransactionWithMeta[]) => Promise<string | undefined | void>,
    batchSize: number,
  ): Promise<string | undefined> {
    let before: string | undefined = undefined;
    const limit = batchSize;
  
    while (true) {
      const signatures = await config.RPC.getSignatures(pubkey, { before, limit });
      if (signatures.length === 0) break;
      before = signatures[signatures.length - 1].signature;

      const filteredSignatures = await this.filterExistingSignatures(signatures.filter(x => !x.err).map(x => x.signature));
      if (filteredSignatures.length === 0) continue;

      const rawTransactions = await config.RPC.getBatchTransactions(filteredSignatures);
      const transactions = rawTransactions.filter((tx): tx is ParsedTransactionWithMeta => tx !== null);
      if (transactions.length === 0) continue;

      const result = await batchProcessor(transactions);

      // note: needed for returning mint from history
      if (result) return result;
    }
  
    console.log(`[getTransactions] Finished processing all transaction batches for account: ${pubkey.toBase58()}`);
    return undefined;
  }

  private async filterExistingSignatures(signatures: string[]): Promise<string[]> {
    const existenceChecks = signatures.map(signature => this.db.signatureExists(signature));
    const existenceResults = await Promise.all(existenceChecks);
    return signatures.filter((_, index) => !existenceResults[index]);
  }
}

Parser:

Parses transaction data and extracts relevant token information.

Detects new token accounts

Computes token account balances and token supply

Saves events and updates balances/supply in the database


export class Parser {
    private db: Database;
    private fetcher!: Fetcher;

    constructor(db: Database) {
        this.db = db;
    }

    public async tokenMovements(transaction: ParsedTransactionWithMeta): Promise<void> {
        await this.parseInstructions(transaction, async (instruction, info) => {
            const { type } = instruction.parsed;
            const signature = transaction.transaction.signatures[0];
            const signers = transaction.transaction.message.accountKeys
                .filter(x => x.signer)
                .map(x => String(x.pubkey));

            switch (true) {
                case type.includes('initializeAccount'):
                    if (!info.mint) return null;
                    await this.handleInitAccount(info, signers, signature);
                    break;

                case type === 'transfer' || type === 'transferChecked':
                    if (!info.mint) info.mint = await this.getMint([info.source, info.destination]);
                    await this.handleTransfer(info, signers, signature);
                    break;

                case type === 'mintTo' || type === 'mintToChecked':
                    await this.handleMint(info, signers, signature);
                    break;

                case type === 'burn' || type === 'burnChecked':
                    await this.handleBurn(info, signers, signature);
                    break;

                default:
                    break;
            }

            return null;
        });
    }

    public async mintFromHistory(transaction: ParsedTransactionWithMeta, account: string): Promise<string | null> {
        return await this.parseInstructions(transaction, async (instruction, info) => {
            if (this.isValidInstruction(instruction, info, account)) {
                await this.db.saveTokenAccount({ 
                    address: account, 
                    mint: config.TOKEN, 
                    owner: info.owner, 
                    balance: '0' 
                });
                return info.mint;
            } 
            
            if (this.isRelatedAccount(info, account) && info.mint !== config.TOKEN) {
                console.log('this account is not from mint', account, info.mint);
                return '';
            }
        
            return null;
        });
    }

    private async parseInstructions(
        transaction: ParsedTransactionWithMeta,
        callback: (instruction: ParsedInstruction, info: any) => Promise<string | null>
    ): Promise<string | null> {
        if (!transaction.meta?.innerInstructions) return null;
        for (const innerInstruction of transaction.meta.innerInstructions) {
            for (const instruction of innerInstruction.instructions) {
                if ('parsed' in instruction && instruction.program === 'spl-token') {
                    const result = await callback(instruction as ParsedInstruction, instruction.parsed.info);
                    if (result) return result;
                }
            }
        }

        return null;
    }

    private isRelatedAccount(info: any, account: string): boolean {
        return info.account === account || info.destination === account || info.source === account;
    }

    private isValidInstruction(instruction: ParsedInstruction, info: any, account: string): boolean {
        return (
            instruction.program === 'spl-token' &&
            info.account === account &&
            info.mint === config.TOKEN &&
            'owner' in info
        );
    }

    public async mint(mint: string, accountInfo: AccountInfo<Buffer> | null): Promise<void> {
        if (!accountInfo || !accountInfo.owner.equals(TOKEN_PROGRAM_ID)) return;
        
        const decodedMintData: Mint = MintLayout.decode(accountInfo.data);
        const mintData: ParsedMint = {
            mint,
            mintAuthorityOption: decodedMintData.mintAuthorityOption,
            mintAuthority: decodedMintData.mintAuthority?.toBase58() || '',
            supply: '0',
            decimals: decodedMintData.decimals,
            isInitialized: decodedMintData.isInitialized,
            freezeAuthorityOption: decodedMintData.freezeAuthorityOption,
            freezeAuthority: decodedMintData.freezeAuthority?.toBase58() || '',
        };

        await this.db.saveMint(mintData);
    }

    private async handleInitAccount(info: any, signers: string[], signature: string): Promise<void> {
        const { account: address, owner, mint } = info;
        if (mint !== config.TOKEN) return;

        const tokenAccount = { address, mint: config.TOKEN, owner };
        if (!await this.db.tokenAccountExists(address)) {
            await this.db.saveTokenAccount({...tokenAccount, balance: '0'});
        }

        await this.db.saveEvent({
            signature,
            type: 'initAccount',
            signers,
            ...tokenAccount
        });
    }

    private async handleTransfer(info: any, signers: string[], signature: string): Promise<void> {
        const { source, destination, amount, mint } = info;
        if (mint !== config.TOKEN) return;

        const amountBN = new BN(amount);
        await Promise.all([
            this.updateBalances(source, amountBN.neg(), destination, amountBN),
            this.db.saveEvent({
                signature,
                type: 'transfer',
                signers,
                ...info,
                amount: bnToHex(amountBN),
            })
        ]);
    }

    private async handleMint(info: any, signers: string[], signature: string): Promise<void> {
        const { mint, account: destination, amount } = info;
        if (mint !== config.TOKEN) return;

        const amountBN = new BN(amount);
        await Promise.all([
            this.updateBalances(null, new BN(0), destination, amountBN),
            this.updateSupply(mint, amountBN),
            this.db.saveEvent({
                signature,
                type: 'mint',
                destination,
                mint,
                amount: bnToHex(amountBN),
                signers,
            })
        ]);
    }

    private async handleBurn(info: any, signers: string[], signature: string): Promise<void> {
        const { mint, account: source, amount } = info;
        if (mint !== config.TOKEN) return;

        const amountBN = new BN(amount);
        await Promise.all([
            this.updateBalances(source, amountBN.neg(), null, new BN(0)),
            this.updateSupply(mint, amountBN.neg()),
            this.db.saveEvent({
                signature,
                type: 'burn',
                source,
                mint,
                amount: bnToHex(amountBN),
                signers,
            })
        ]);
    }

    private async updateBalances(fromAddress: string | null, fromAmount: BN, toAddress: string | null, toAmount: BN): Promise<void> {
        if (fromAddress) {
            const fromBalance = await this.getBalance(fromAddress);
            const newFromBalance = fromBalance.add(fromAmount);
            await this.db.updateBalance(fromAddress, bnToHex(newFromBalance));
        }

        if (toAddress) {
            const toBalance = await this.getBalance(toAddress);
            const newToBalance = toBalance.add(toAmount);
            await this.db.updateBalance(toAddress, bnToHex(newToBalance));
        }
    }

    private async updateSupply(mintAddress: string, amount: BN): Promise<void> {
        const currentSupply = await this.getSupply(mintAddress);
        const newSupply = currentSupply.add(amount);
        await this.db.updateTokenSupply(mintAddress, bnToHex(newSupply));
    }

    private async getMint(addresses: string[]): Promise<string> {
        const mint = await this.db.mintFromAccounts(addresses);
        return mint ? mint : await this.fetcher.mintFromHistory(addresses)
    }

    private async getBalance(address: string): Promise<BN> {
        const tokenAccount = await this.db.getTokenAccount(address);
        const hexBalance = tokenAccount?.balance;
        return hexBalance ? hexToBN(hexBalance) : new BN(0);
    }

    private async getSupply(mintAddress: string): Promise<BN> {
        const hexSupply = await this.db.getTokenSupply(mintAddress);
        return hexSupply ? hexToBN(hexSupply) : new BN(0);
    }
}

Webhook listener:

Receives new transactions, waits until they are confirmed, and sends them to the parser. You need to set up a server where this code is deployed and the webhook on your RPC provider.


export class Listener {
    private parser: Parser;
    private app: Elysia;

    constructor(parser: Parser) {
        this.parser = parser;
        this.app = new Elysia();
        this.setupRoutes();
    }

    private setupRoutes() {
        this.app.post('/programListener', async ({ body, headers }) => {
            try {
                const authToken = headers['authorization'];
                if (!authToken || authToken !== config.RPC_KEY) {
                    console.error(`Unauthorized request`);
                    return new Response(JSON.stringify({ error: "Unauthorized" }), {
                        status: 401,
                        headers: { 'Content-Type': 'application/json' }
                    });
                }

                const signatures = (body as any).flatMap((x: any) => x.transaction.signatures);
                const confirmationPromises = signatures.map((signature: string) => config.RPC.getConfirmation(signature));
                const confirmationResults = await Promise.all(confirmationPromises);
                const confirmedSignatures = signatures.filter((_: string, index: number) => confirmationResults[index] !== null);
            
                if (confirmedSignatures.length === 0) {
                  console.log('No transactions were confirmed');
                  return { success: false, message: 'No transactions were confirmed' };
                }
            
                console.log(`Confirmed signatures: ${confirmedSignatures}`);
                const rawTransactions = await config.RPC.getBatchTransactions(signatures);
                const transactions = rawTransactions.filter((tx): tx is ParsedTransactionWithMeta => tx !== null);
                for (const transaction of transactions) {
                    await this.parser.tokenMovements(transaction);
                }
                
                return { success: true, message: 'Transactions processed successfully' };
            } catch (error) {
                console.error('Failed to process transactions:', error);
                return { success: false, message: 'Failed to process transactions' };
            }
        });
    }

    public getHandler() {
        return this.app;
    }
}

Database:

Using firebase for simplicity, really easy to setup and to visualize data, I recommend using sqlite bun plugin for tokens with a lot of activity, for this iteration I used real time db from firebase but in the future will use that for sure.


import { getDatabase } from "firebase-admin/database";
import admin from "firebase-admin";
import { config } from "./config";
import { type ParsedMint, type TokenAccount } from "./types";

const app = admin.apps.find((it: any) => it?.name === "[DEFAULT]") ||
    admin.initializeApp({
        credential: admin.credential.cert({
            projectId: config.FIREBASE_PROJECT_ID,
            clientEmail: config.FIREBASE_CLIENT_EMAIL,
            privateKey: config.FIREBASE_PRIVATE_KEY!.replace(/\\n/gm, "\n"),
        }),
        databaseURL: config.FIREBASE_DATABASE
    });

const database = getDatabase(app);

export class Database {
    private eventsRef = database.ref('events');
    private mintsRef = database.ref('mints');
    private tokenAccountsRef = database.ref('tokenAccounts');

    public async signatureExists(signature: string): Promise<boolean> {
        const snapshot = await this.eventsRef.child('signatures').child(signature).once('value');
        return snapshot.exists();
    }

    public async tokenAccountExists(address: string): Promise<boolean> {
        const snapshot = await this.tokenAccountsRef.child(address).once('value');
        return snapshot.exists();
    }

    public async mintFromAccounts(accounts: string[]): Promise<string | null> {
        for (const account of accounts) {
            const tokenAccount = await this.getTokenAccount(account);
            if (tokenAccount && tokenAccount.mint) {
                const mint = await this.getMint(tokenAccount.mint);
                if (mint) {
                    return mint.mint;
                }
            }
        }
    
        return null;
    }

    public async saveEvent(event: any): Promise<void> {
        const updates: { [key: string]: any } = {};
        updates[`${event.type}/${event.signature}`] = event;
        updates[`signatures/${event.signature}`] = true;
        await this.eventsRef.update(updates);
    }

    public async saveMint(mintData: ParsedMint): Promise<void> {
        await this.mintsRef.child(mintData.mint).set(mintData);
    }

    public async saveTokenAccount(tokenAccount: TokenAccount): Promise<void> {
        await this.tokenAccountsRef.child(tokenAccount.address).set(tokenAccount);
    }

    public async updateBalance(address: string, balance: string): Promise<void> {
        await this.tokenAccountsRef.child(address).update({ balance });
    }

    public async updateTokenSupply(mintAddress: string, supply: string): Promise<void> {
        await this.mintsRef.child(mintAddress).update({ supply });
    }

    public async getMint(mintAddress: string): Promise<ParsedMint | null> {
        const snapshot = await this.mintsRef.child(mintAddress).once('value');
        return snapshot.exists() ? snapshot.val() as ParsedMint : null;
    }

    public async getBalance(address: string): Promise<string | null> {
        const snapshot = await this.tokenAccountsRef.child(address).child('balance').once('value');
        return snapshot.val();
    }

    public async getAllBalances(): Promise<Map<string, string>> {
        const snapshot = await this.tokenAccountsRef.once('value');
        const balances = new Map<string, string>();
        snapshot.forEach((childSnapshot) => {
            const tokenAccount = childSnapshot.val();
            if (tokenAccount.balance) {
                balances.set(childSnapshot.key!, tokenAccount.balance);
            }
        });
        return balances;
    }

    public async getTokenSupply(mintAddress: string): Promise<string | null> {
        const mint = await this.getMint(mintAddress);
        return mint ? mint.supply : null;
    }

    public async getTokenAccount(address: string): Promise<any | null> {
        const snapshot = await this.tokenAccountsRef.child(address).once('value');
        return snapshot.exists() ? snapshot.val() : null;
    }
}

Server Setup

Install Docker on the server (Ubuntu):

Follow the official Docker installation guide for Ubuntu

Create a docker-compose.yml file:


services:
  token-monitor:
    image: ricardocr987/token-monitor:latest
    env_file: .env
    restart: unless-stopped
    volumes:
      - ./logs:/app/logs
    ports:
      - "3001:3001"
    networks:
      - token-network

volumes:
  logs:

networks:
  token-network:
    name: token-network
    driver: bridge

Configure firewall and start Docker:


sudo systemctl start docker
sudo systemctl enable docker
sudo ufw allow ssh 3001
sudo ufw enable

Updating the Docker Container

On local machine:


docker build -t solana-server .
docker tag solana-server:latest ricardocr987/solana-server:latest
docker push ricardocr987/solana-server

On server:


docker stop token-monitor-token-monitor-1
docker rm token-monitor-token-monitor-1
docker pull ricardocr987/mango-server:latest
docker compose up --build -d
docker logs token-monitor-token-monitor-1

Troubleshooting

If you encounter issues with Docker installation on Ubuntu, follow these steps:

Determine the correct Ubuntu codename:


cat /etc/os-release | grep -i codename

Update the /etc/apt/sources.list.d/docker.list file with the correct codename:


https://download.docker.com/linux/ubuntu <CORRECT_CODENAME> stable

Token indexing