Copy import * as sb from '@switchboard-xyz/on-demand';
import { SuiClient } from '@mysten/sui/client';
import {
SwitchboardClient,
emitSurgeQuote,
} from '@switchboard-xyz/sui-sdk';
import { fromB64 } from '@mysten/bcs';
import { Ed25519Keypair } from '@mysten/sui/keypairs/ed25519';
import * as path from 'path';
import * as os from 'os';
import * as fs from 'fs';
import { Transaction } from '@mysten/sui/transactions';
// Initialize Sui clients
const suiClient = new SuiClient({ url: 'https://fullnode.mainnet.sui.io:443' });
const switchboardClient = new SwitchboardClient(suiClient);
// Oracle mapping cache
const oracleMapping = new Map<string, string>();
let lastOracleFetch = 0;
const ORACLE_CACHE_TTL = 1000 * 60 * 10; // 10 minutes
// Transaction queue management
let isTransactionProcessing = false;
const rawResponseQueue: Array<{
rawResponse: any;
timestamp: number;
}> = [];
// Process transaction queue - ensures only one transaction at a time
async function processTransactionQueue(): Promise<void> {
if (isTransactionProcessing || rawResponseQueue.length === 0) {
return;
}
isTransactionProcessing = true;
try {
const queueItem = rawResponseQueue.shift()!;
const { rawResponse, timestamp } = queueItem;
console.log(`Processing transaction (queue length: ${rawResponseQueue.length})`);
const transaction = new Transaction();
// Convert Surge update to Sui transaction
await emitSurgeQuote(switchboardClient, transaction, rawResponse);
const result = await suiClient.signAndExecuteTransaction({
transaction: transaction,
signer: keypair!,
options: {
showEvents: true,
showEffects: true,
},
});
const processingTime = Date.now() - timestamp;
console.log(`Transaction completed in ${processingTime}ms`);
console.log('Transaction result:', result.digest);
} catch (error) {
console.error('Transaction failed:', error);
} finally {
isTransactionProcessing = false;
// Process next transaction in queue if any
if (rawResponseQueue.length > 0) {
setImmediate(() => processTransactionQueue());
}
}
}
// Fetch oracle mappings from Crossbar
async function fetchOracleMappings(): Promise<Map<string, string>> {
const now = Date.now();
if (oracleMapping.size > 0 && now - lastOracleFetch < ORACLE_CACHE_TTL) {
return oracleMapping;
}
try {
const response = await fetch('https://crossbar.switchboard.xyz/oracles/sui');
const oracles = (await response.json()) as Array<{
oracle_id: string;
oracle_key: string;
}>;
oracleMapping.clear();
for (const oracle of oracles) {
const cleanKey = oracle.oracle_key.startsWith('0x')
? oracle.oracle_key.slice(2)
: oracle.oracle_key;
oracleMapping.set(cleanKey, oracle.oracle_id);
}
lastOracleFetch = now;
console.log(`Loaded ${oracleMapping.size} oracle mappings`);
return oracleMapping;
} catch (error) {
console.error('Failed to fetch oracle mappings:', error);
return oracleMapping;
}
}
// Calculate latency statistics
function calculateStatistics(latencies: number[]) {
const sorted = [...latencies].sort((a, b) => a - b);
const sum = sorted.reduce((a, b) => a + b, 0);
return {
min: sorted[0],
max: sorted[sorted.length - 1],
median: sorted[Math.floor(sorted.length / 2)],
mean: sum / sorted.length,
count: sorted.length,
};
}
// Load keypair from Sui keystore
let keypair: Ed25519Keypair | null = null;
try {
const keystorePath = path.join(os.homedir(), '.sui', 'sui_config', 'sui.keystore');
const keystore = JSON.parse(fs.readFileSync(keystorePath, 'utf-8'));
const secretKey = fromB64(keystore[0]);
keypair = Ed25519Keypair.fromSecretKey(secretKey.slice(1));
} catch (error) {
console.error('Error loading keypair:', error);
}
if (!keypair) {
throw new Error('Keypair not loaded');
}
// Main function
(async function main() {
console.log('Starting Surge streaming...');
console.log(`Using keypair: ${keypair!.toSuiAddress()}`);
const latencies: number[] = [];
// Initialize Surge with keypair and connection (uses on-chain subscription)
const surge = new sb.Surge({
connection: suiClient,
keypair: keypair!,
signatureScheme: 'secp256k1',
});
// Connect and subscribe to feeds
await surge.connectAndSubscribe([{ symbol: 'BTC/USD' }]);
// Pre-fetch oracle mappings
await fetchOracleMappings();
// Listen for price updates
surge.on('signedPriceUpdate', async (response: sb.SurgeUpdate) => {
const currentLatency = Date.now() - response.data.source_ts_ms;
latencies.push(currentLatency);
const rawResponse = response.getRawResponse();
const stats = calculateStatistics(latencies);
const formattedPrices = response.getFormattedPrices();
const currentPrice = Object.values(formattedPrices)[0] || 'N/A';
console.log(
`Update #${stats.count} | Price: ${currentPrice} | Latency: ${currentLatency}ms | Avg: ${stats.mean.toFixed(1)}ms`
);
// Queue the update for processing
rawResponseQueue.push({
rawResponse,
timestamp: Date.now(),
});
// Trigger queue processing
processTransactionQueue();
});
console.log('Listening for price updates...');
})(); 