Production-Ready Quote Service & Scanner Framework: Oracle-Based Quotes and Extensible Architecture

// ❌ WRONG - Meaningless reverse amounts
Forward:  SOL → USDC with 1 SOL (1000000000 lamports)
Reverse:  USDC → SOL with 1 SOL (1000000000 lamports)
// This means "swap 0.000000001 USDC for SOL" - useless!

// ✅ CORRECT - Oracle-calculated equivalent amounts
Forward:  SOL → USDC with 1 SOL (1000000000 lamports)
Reverse:  USDC → SOL with 140 USDC (140000000 lamports)
// At $140/SOL: economically equivalent round-trip

// ❌ BEFORE - Requesting 16 pairs for 8 LST tokens (redundant!)
for (const lst of LST_TOKENS) {
    pairs.push({ inputMint: sol.mint, outputMint: lst.mint });  // SOL → LST
    pairs.push({ inputMint: lst.mint, outputMint: sol.mint });  // LST → SOL (redundant!)
}

// Total gRPC subscriptions: 16 pairs × 29 amounts = 464 combinations

// ✅ AFTER - Request only forward pairs, quote-service auto-adds reverse
for (const lst of LST_TOKENS) {
    pairs.push({ inputMint: sol.mint, outputMint: lst.mint });  // SOL → LST only
    // Go service automatically creates: LST → SOL with oracle-calculated amount
}

// Total gRPC subscriptions: 8 pairs × 29 amounts = 232 combinations
// Network load: 50% reduction
// Quote cache: 232 forward + 232 reverse (auto-added) = 464 quotes

// ❌ OLD - Exact amount matching (breaks with dynamic oracle)
const reverseCacheKey = this.getCacheKey(
    quote.outputMint,
    quote.inputMint,
    quote.outAmount  // Exact match required
);
const reverseQuote = this.quoteCache.get(reverseCacheKey);

// ✅ NEW - Flexible amount matching
private findReverseQuote(reversePairKey: string, targetAmount: string): QuoteData | undefined {
    const target = BigInt(targetAmount);
    const tolerance = target / 10n; // ±10% tolerance

    // Search cache for matching pair with similar amount
    for (const [key, data] of this.quoteCache.entries()) {
        if (key.startsWith(reversePairKey)) {
            const amount = BigInt(data.quote.inAmount);
            const diff = amount > target ? amount - target : target - amount;

            // Match if within tolerance
            if (diff <= tolerance) {
                return data;
            }
        }
    }

    return undefined;
}

// ❌ OLD - Custom scanner with manual infrastructure
class ArbitrageScannerService {
    private natsClient: NatsConnection;
    private redis: Redis;
    private grpcClient: GrpcQuoteClient;
    private handler: QuoteStreamHandler;

    async start() {
        // Manual NATS connection (50 lines)
        this.natsClient = await connect({ servers: [...] });

        // Manual Redis connection (40 lines)
        this.redis = new Redis({ ... });

        // Manual metrics setup (60 lines)
        this.setupMetrics();

        // Manual error handling (50 lines)
        this.setupErrorHandlers();

        // Manual event loop (100+ lines)
        await this.runEventLoop();
    }
}

// ✅ NEW - Framework-based scanner
export class ArbitrageQuoteScanner extends SubscriptionScanner {
    name = 'arbitrage-quote';
    maxConcurrency = 10;
    maxQueueSize = 1000;

    private grpcClient: GrpcQuoteClient;
    private handler: QuoteStreamHandler;

    constructor(config: SubscriptionScannerConfig, arbitrageConfig: ArbitrageConfig) {
        super(config);

        this.grpcClient = new GrpcQuoteClient(
            arbitrageConfig.grpc.host,
            arbitrageConfig.grpc.port
        );

        this.handler = new QuoteStreamHandler(
            arbitrageConfig.tokenPairs,
            arbitrageConfig.amounts,
            arbitrageConfig.minProfitBps,
            arbitrageConfig.maxSlippageBps,
            arbitrageConfig.minConfidence
        );
    }

    // Subscribe to gRPC stream and return async iterator
    protected async subscribe(): Promise<AsyncIterator<any>> {
        await this.grpcClient.connect();

        // Subscribe to token pairs
        await this.grpcClient.subscribeToPairs(
            this.config.customConfig.tokenPairs,
            this.config.customConfig.amounts.map(a => a.toString()),
            this.config.customConfig.maxSlippageBps
        );

        // Convert EventEmitter to AsyncIterator
        return this.createQuoteIterator();
    }

    // Process each quote and detect opportunities
    protected async process(quote: any): Promise<MarketEvent[]> {
        const events: MarketEvent[] = [];

        // Process quote through handler
        this.handler.handleQuote(quote);

        // Wait for opportunity detection (100ms timeout)
        const opportunity = await this.waitForOpportunity(100);

        if (opportunity) {
            // Validate opportunity
            if (opportunity.confidence >= this.config.customConfig.minConfidence) {
                events.push({
                    type: 'ArbitrageOpportunity',
                    ...opportunity,
                    timestamp: Date.now(),
                    slot: quote.contextSlot || 0,
                    metadata: {
                        provider: quote.provider,
                        slippageBps: quote.slippageBps,
                    },
                });

                // Update custom stats
                this.stats.customStats.opportunitiesDetected =
                    (this.stats.customStats.opportunitiesDetected || 0) + 1;
            }
        }

        return events;
    }

    // Generate unique event ID for deduplication
    protected getEventId(event: MarketEvent): string {
        if (event.type === 'ArbitrageOpportunity') {
            return event.opportunityId;
        }
        return `${event.type}:${event.timestamp}`;
    }

    // Convert gRPC EventEmitter to AsyncIterator
    private async *createQuoteIterator(): AsyncIterator<any> {
        const queue: any[] = [];
        let resolveNext: ((value: any) => void) | null = null;

        this.grpcClient.on('quote', (quote) => {
            if (resolveNext) {
                resolveNext(quote);
                resolveNext = null;
            } else {
                queue.push(quote);
            }
        });

        while (this.running) {
            if (queue.length > 0) {
                yield queue.shift();
            } else {
                const quote = await new Promise<any>((resolve) => {
                    resolveNext = resolve;
                    setTimeout(() => {
                        if (resolveNext === resolve) {
                            resolveNext = null;
                            resolve(null);
                        }
                    }, 30000);
                });

                if (quote) {
                    yield quote;
                }
            }
        }
    }
}

// Automatic event publishing with retry
protected async publish(event: MarketEvent): Promise<void> {
    const subject = this.getNatsSubject(event);
    await this.jsClient.publish(subject, JSON.stringify(event));
    this.metrics.incrementEventsPublished(event.type);
}

// Automatic duplicate detection
protected async isDuplicate(event: MarketEvent): Promise<boolean> {
    const key = `dedup:${this.name()}:${this.getEventId(event)}`;
    const exists = await this.redis.exists(key);

    if (!exists) {
        await this.redis.setex(key, this.config.deduplicationTtl, '1');
        return false;
    }

    return true;
}

// Automatic metric collection
scanner_events_published_total{scanner="arbitrage-quote", event_type="ArbitrageOpportunity"}
scanner_errors_total{scanner="arbitrage-quote", error_type="grpc_error"}
scanner_processing_duration_seconds{scanner="arbitrage-quote"}
scanner_queue_size{scanner="arbitrage-quote"}
scanner_queue_overflows_total{scanner="arbitrage-quote"}

// Automatic startup/shutdown
await scanner.start();
// - Connects to NATS
// - Connects to Redis
// - Starts event processing
// - Registers shutdown handlers

await scanner.stop();
// - Drains processing queue
// - Closes connections
// - Publishes final metrics

import { PollingScanner, type PollingScannerConfig } from '@repo/scanner-framework';
import type { MarketEvent } from '@repo/market-events';
import { logger } from '@repo/observability';

interface DCAConfig {
    tokenMint: string;      // Token to buy
    quoteMint: string;      // Token to spend
    interval: number;       // Seconds between executions
    amount: number;         // Amount to spend (smallest units)
    targetPrice?: number;   // Optional price limit
    pairName?: string;      // Human-readable name
}

export class DCAScanner extends PollingScanner {
    name = 'dca-scanner';
    pollInterval: number;

    private dcaConfig: DCAConfig;
    private priceClient: PriceClient;
    private lastExecutionTime: number = 0;

    constructor(config: PollingScannerConfig, dcaConfig: DCAConfig) {
        super(config);
        this.dcaConfig = dcaConfig;
        this.pollInterval = dcaConfig.interval * 1000;
        this.priceClient = new PriceClient();
    }

    // Fetch current price
    protected async fetch(): Promise<any> {
        const price = await this.priceClient.getPrice(
            this.dcaConfig.tokenMint,
            this.dcaConfig.quoteMint
        );

        return {
            price,
            timestamp: Date.now(),
        };
    }

    // Check conditions and trigger DCA order
    protected async process(data: any): Promise<MarketEvent[]> {
        const events: MarketEvent[] = [];
        const now = Date.now();

        // Check if interval has passed
        const intervalPassed = (now - this.lastExecutionTime) >= (this.dcaConfig.interval * 1000);

        if (!intervalPassed) {
            this.stats.customStats.skippedDueToInterval =
                (this.stats.customStats.skippedDueToInterval || 0) + 1;
            return events;
        }

        // Check if price condition is met (optional)
        if (this.dcaConfig.targetPrice) {
            const targetMet = data.price <= this.dcaConfig.targetPrice;
            if (!targetMet) {
                logger.debug({
                    scanner: this.name,
                    currentPrice: data.price,
                    targetPrice: this.dcaConfig.targetPrice,
                }, 'DCA target price not met');

                this.stats.customStats.skippedDueToPrice =
                    (this.stats.customStats.skippedDueToPrice || 0) + 1;
                return events;
            }
        }

        // Create DCA execution event
        const expectedOutput = Math.floor(this.dcaConfig.amount / data.price);

        events.push({
            type: 'SwapRoute',
            routeId: `dca:${this.dcaConfig.pairName}:interval=${this.dcaConfig.interval}:price=${data.price}:${now}`,
            tokenIn: this.dcaConfig.quoteMint,
            tokenOut: this.dcaConfig.tokenMint,
            amountIn: this.dcaConfig.amount.toString(),
            expectedAmountOut: expectedOutput.toString(),
            hops: [{
                label: 'DCA',
                ammKey: 'dca-pool',
                inputMint: this.dcaConfig.quoteMint,
                outputMint: this.dcaConfig.tokenMint,
                inAmount: this.dcaConfig.amount.toString(),
                outAmount: expectedOutput.toString(),
            }],
            timestamp: now,
            slot: 0,
            metadata: {
                dcaInterval: this.dcaConfig.interval,
                targetPrice: this.dcaConfig.targetPrice,
                currentPrice: data.price,
            },
        });

        this.lastExecutionTime = now;

        // Update custom stats
        this.stats.customStats.dcaExecutions =
            (this.stats.customStats.dcaExecutions || 0) + 1;
        this.stats.customStats.lastExecutionPrice = data.price;
        this.stats.customStats.lastExecutionTime = now;

        logger.info({
            scanner: this.name,
            tokenIn: this.dcaConfig.quoteMint,
            tokenOut: this.dcaConfig.tokenMint,
            amount: this.dcaConfig.amount,
            price: data.price,
        }, 'DCA execution triggered');

        return events;
    }

    // Don't deduplicate DCA events - each is unique
    protected getEventId(event: MarketEvent): string {
        return `dca:${event.timestamp}:${Math.random()}`;
    }
}

// Configure DCA: Buy 100 USDC worth of SOL every hour if price <= $100
const dcaScanner = new DCAScanner(
    {
        name: 'dca-sol-usdc',
        enabled: true,
        type: ScannerType.Polling,
        natsServers: ['nats://localhost:4222'],
        redisUrl: 'redis://localhost:6379',
        deduplicationTtl: 5,
        maxRetries: 3,
        metricsEnabled: true,
        deduplicationEnabled: false, // Each DCA execution is unique
        customConfig: {},
        pollInterval: 3600000, // Overridden by DCAConfig
    },
    {
        tokenMint: 'So11111111111111111111111111111111111111112', // SOL
        quoteMint: 'EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v', // USDC
        interval: 3600,          // 1 hour
        amount: 100_000_000,     // 100 USDC (6 decimals)
        targetPrice: 100,        // Only buy if SOL <= $100
        pairName: 'SOL/USDC',
    }
);

await dcaScanner.start();

export class PriceMonitorScanner extends PollingScanner {
    name = 'price-monitor';
    pollInterval = 10000; // 10 seconds

    protected async fetch(): Promise<any> {
        // Fetch from price API
        return await this.priceApi.getPrices();
    }

    protected async process(data: any): Promise<MarketEvent[]> {
        // Convert to market events
        return data.map(price => ({
            type: 'PriceUpdate',
            token: price.mint,
            price: price.usd,
            timestamp: Date.now(),
            slot: 0,
            metadata: {},
        }));
    }

    protected getEventId(event: MarketEvent): string {
        return `${event.token}:${event.timestamp}`;
    }
}

export class PoolUpdateScanner extends SubscriptionScanner {
    name = 'pool-update';
    maxConcurrency = 20;
    maxQueueSize = 2000;

    private shredstream: ShredstreamClient;

    protected async subscribe(): Promise<AsyncIterator<any>> {
        await this.shredstream.connect();
        return this.shredstream.subscribeToAccounts(POOL_ADDRESSES);
    }

    protected async process(update: any): Promise<MarketEvent[]> {
        // Parse pool account data
        const poolData = parsePoolAccount(update.account);

        return [{
            type: 'PoolUpdate',
            poolAddress: update.pubkey,
            tokenA: poolData.mintA,
            tokenB: poolData.mintB,
            reserveA: poolData.reserveA,
            reserveB: poolData.reserveB,
            slot: update.slot,
            timestamp: Date.now(),
            metadata: {
                protocol: poolData.protocol,
            },
        }];
    }

    protected getEventId(event: MarketEvent): string {
        return `${event.poolAddress}:${event.slot}`;
    }
}

export class MultiSourceScanner extends SubscriptionScanner {
    name = 'multi-source';
    maxConcurrency = 20;
    maxQueueSize = 2000;

    private sourceA: SourceAClient;
    private sourceB: SourceBClient;
    private updateQueue: any[] = [];

    protected async subscribe(): Promise<AsyncIterator<any>> {
        // Connect both sources
        await Promise.all([
            this.sourceA.connect(),
            this.sourceB.connect(),
        ]);

        // Merge streams
        this.sourceA.on('data', (data) => this.enqueueUpdate({ source: 'A', data }));
        this.sourceB.on('data', (data) => this.enqueueUpdate({ source: 'B', data }));

        return this.createIterator();
    }

    protected async process(update: any): Promise<MarketEvent[]> {
        if (update.source === 'A') {
            return this.processSourceA(update.data);
        } else {
            return this.processSourceB(update.data);
        }
    }

    // ... iterator and processing logic
}

export class HandlerBasedScanner extends SubscriptionScanner {
    name = 'handler-scanner';
    maxConcurrency = 10;
    maxQueueSize = 1000;

    private dataClient: DataClient;
    private handler: ProcessingHandler;

    protected async subscribe(): Promise<AsyncIterator<any>> {
        await this.dataClient.connect();
        return this.dataClient.stream();
    }

    protected async process(data: any): Promise<MarketEvent[]> {
        // Delegate to handler
        await this.handler.process(data);

        // Wait for result (with timeout)
        const result = await this.waitForResult(100);

        if (result) {
            return [{
                type: 'HandlerResult',
                data: result,
                timestamp: Date.now(),
                slot: 0,
                metadata: {},
            }];
        }

        return [];
    }

    protected getEventId(event: MarketEvent): string {
        return `${event.type}:${event.timestamp}`;
    }
}

private async *createQuoteIterator(): AsyncIterator<any> {
    const queue: any[] = [];
    let resolveNext: ((value: any) => void) | null = null;

    // EventEmitter → Queue
    this.grpcClient.on('quote', (quote) => {
        if (resolveNext) {
            resolveNext(quote);
            resolveNext = null;
        } else {
            queue.push(quote);
        }
    });

    // Queue → AsyncIterator
    while (this.running) {
        if (queue.length > 0) {
            yield queue.shift();
        } else {
            const quote = await new Promise<any>((resolve) => {
                resolveNext = resolve;
                setTimeout(() => {
                    if (resolveNext === resolve) {
                        resolveNext = null;
                        resolve(null);
                    }
                }, 30000); // Timeout after 30s
            });

            if (quote) {
                yield quote;
            }
        }
    }
}

// ❌ WRONG
import { incrementCounter, observeHistogram } from '@repo/observability';

// ✅ CORRECT
import { metrics } from '@repo/observability';

metrics.incrementCounter('scanner_events_published_total', { scanner: 'my-scanner' });
metrics.observeHistogram('scanner_poll_duration_seconds', durationSec, { scanner: 'my-scanner' });
metrics.setGauge('scanner_queue_size', queueSize, { scanner: 'my-scanner' });