Skip to content

Instantly share code, notes, and snippets.

@Leeyah-123

Leeyah-123/stream-filter.js

Last active January 12, 2025 20:50
Show Gist options
  • Save Leeyah-123/98031b6fbf39b1fb2570c92e1a7b2df1 to your computer and use it in GitHub Desktop.
Save Leeyah-123/98031b6fbf39b1fb2570c92e1a7b2df1 to your computer and use it in GitHub Desktop.
Download ZIP
QuickNode stream filter to filter out transactions we don't need and function to transform QuickNode Solana Programs+Logs Stream data to human-readable format
Raw
stream-filter.js
function main(stream) {
const data = Array.isArray(stream.data) ? stream.data[0] : stream.data;
// Obtain addresses from KV Store
const addresses = qnGetList("watchlist_addresses");
// Convert addresses array to Set for faster lookups
const addressSet = new Set(addresses?.map(addr => addr.toLowerCase()) || []);
const filteredData = data.filter(tx => {
// Check all accounts in all program invocations
return tx.programInvocations?.some(inv =>
inv.instruction.accounts.some(acc =>
addressSet.has(acc.pubkey.toLowerCase())
)
);
});
console.log(`Reduced data from ${data.length} tx(s) to ${filteredData.length} tx(s).`)
return filteredData;
}
Raw
transform-solana-programs-and-logs-function.js
// Import dependencies, learn more at: https://www.quicknode.com/docs/functions/runtimes/node-js-20-runtime
const axios = require('axios');
const { format } = require('date-fns');
const WEBHOOK_URL = "https://katlog.onrender.com/api/webhook";
const SYSTEM_PROGRAM_ID = '11111111111111111111111111111111';
const VOTE_PROGRAM_ID = 'Vote111111111111111111111111111111111111111';
const TOKEN_PROGRAM_ID = 'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA';
const TOKEN_2022_PROGRAM_ID = 'TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb';
/**
* Parses Solana Programs+Logs into a human-readable format
* @param {Array} data - The Solana Programs+Logs data
*/
const processSolanaTransactions = (data) => {
return data
// Filter out Vote transactions
.filter(tx => !tx.programInvocations?.some(inv =>
inv.programId === VOTE_PROGRAM_ID))
// Map to desired format
.map(tx => {
const programInvocations = tx.programInvocations[0];
// Get all account changes from program invocations
const accounts = programInvocations.instruction.accounts.filter(acc =>
acc.preBalance !== acc.postBalance);
// Format timestamp
const date = new Date(tx.blockTime * 1000);
const formattedDate = format(date, 'MMM d, yyyy HH:mm:ss');
// Find the main from/to accounts based on balance changes
let from, to;
// Determine transaction type based on program ID
let type = 'unknown';
const programId = programInvocations?.programId;
const instruction = programInvocations?.instruction
switch (programId) {
case SYSTEM_PROGRAM_ID:
type = 'SOL Transfer';
break;
case TOKEN_PROGRAM_ID || TOKEN_2022_PROGRAM_ID:
type = 'Token Transfer';
break;
default:
type = 'Other';
}
let amount;
if (type === 'SOL Transfer') {
from = instruction.data.info.source;
to = instruction.data.info.destination;
// For native SOL transfers
amount = Math.abs(instruction.data.info.lamports) / 1e9; // Convert lamports to SOL
} else if (type === 'Token Transfer') {
from = accounts.find(acc => acc.preBalance > acc.postBalance)?.pubkey || '';
to = accounts.find(acc => acc.postBalance > acc.preBalance)?.pubkey || '';
// For token transfers, check token balances
const tokenTransfer = programInvocations?.find(inv =>
inv.instruction.tokenBalances?.some(tb =>
tb.uiTokenAmount?.uiAmount > 0
)
);
if (tokenTransfer?.instruction.tokenBalances) {
const tokenAmount = tokenTransfer.instruction.tokenBalances.find(tb =>
tb.uiTokenAmount?.uiAmount > 0
);
amount = tokenAmount?.uiTokenAmount?.uiAmount || 0;
}
}
return {
signature: tx.signature,
programId,
type,
from,
to,
amount,
success: tx.success,
timestamp: formattedDate,
};
});
};
/**
* Sends data to the specified webhook
* @param {Object} data - The data to send to the webhook
*/
async function sendToWebhook(data) {
try {
const response = await axios.post(WEBHOOK_URL, data, {
headers: {
'Content-Type': 'application/json',
},
});
console.log('Data sent to webhook successfully');
return response.data;
} catch (error) {
if (axios.isAxiosError(error)) {
console.error('Error sending data to webhook:', error.response?.data || error.message);
} else {
console.error('Error sending data to webhook:', error);
}
throw error;
}
}
/**
* main(params) is invoked when your Function is called from Streams or API.
*
* @param {Object} params - May contain a dataset (params.data), params.metadata, and params.user_data
*
* @returns {Object} - A message that will be logged and returned when a Function is called via API.
* Tip: You can chain Functions together by calling them inside each other.
* Learn more at: https://www.quicknode.com/docs/functions/getting-started#overview
*/
async function main(data) {
const parsedData = processSolanaTransactions(data);
if (parsedData.length > 0) {
// Prepare payload for webhook
const payload = JSON.stringify({
data: parsedData,
});
sendToWebhook(payload)
.then(() => console.log(`Sent batch of ${parsedData.length} txs to webhook`))
.catch(err => console.log("Unable to send txs to webhook: ", err));
}
// Return anything that you will consume on API side or helping you check your execution later
return {
message: `Processed batch of ${parsedData.length} txs`,
processedTransactions: parsedData.length,
successfulTransactions: parsedData.reduce((currentValue, { success }) => {
if (success) currentValue++;
return currentValue;
}, 0)
};
}
// Find more examples at https://github.com/quiknode-labs/awesome-functions/
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment