boki/stock-bot
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

refactored out proxymanager from webshare to make it reusable

Browse source
This commit is contained in:
Boki 2025-06-20 11:54:59 -04:00
parent 98aa414231
commit 84cb14680b
8 changed files with 622 additions and 109 deletions
Download patch file Download diff file Expand all files Collapse all files

6
apps/data-service/src/index.ts
View file

@ -114,6 +114,12 @@ async function initializeServices() {
queueManager = QueueManager.getOrInitialize(queueManagerConfig);
logger.info('Queue system initialized');
// Initialize proxy manager
logger.debug('Initializing proxy manager...');
const { proxyManager } = await import('@stock-bot/utils');
await proxyManager.initialize();
logger.info('Proxy manager initialized');
// Initialize providers (register handlers and scheduled jobs)
logger.debug('Initializing data providers...');
const { initializeExchangeSyncProvider } = await import('./providers/exchange-sync.provider');

14
apps/data-service/src/providers/qm.tasks.ts
View file

@ -2,7 +2,7 @@ import { getRandomUserAgent } from '@stock-bot/http';
import { getLogger } from '@stock-bot/logger';
import { getMongoDBClient } from '@stock-bot/mongodb-client';
import { QueueManager } from '@stock-bot/queue';
import { getProxy } from './webshare.provider';
import { proxyManager } from '@stock-bot/utils';
// Shared instances (module-scoped, not global)
let isInitialized = false; // Track if resources are initialized
@ -90,11 +90,15 @@ export async function createSessions(): Promise<void> {
while (sessionCache[sessionId].length < 50) {
logger.info(`Creating new session for ${sessionId}`);
const proxy = getProxy();
if (proxy === null) {
const proxyInfo = await proxyManager.getRandomProxy();
if (!proxyInfo) {
logger.error('No proxy available for QM session creation');
break; // Skip session creation if no proxy is available
}
// Convert ProxyInfo to string format
const auth = proxyInfo.username && proxyInfo.password ? `${proxyInfo.username}:${proxyInfo.password}@` : '';
const proxy = `${proxyInfo.protocol}://${auth}${proxyInfo.host}:${proxyInfo.port}`;
const newSession: QMSession = {
proxy: proxy, // Placeholder, should be set to a valid proxy
headers: getQmHeaders(),
@ -290,9 +294,9 @@ async function searchAndSpawnJobs(
// API call function to search symbols via QM
async function searchQMSymbolsAPI(query: string): Promise<string[]> {
const proxy = getProxy();
const proxyInfo = await proxyManager.getRandomProxy();
if (!proxy) {
if (!proxyInfo) {
throw new Error('No proxy available for QM API call');
}
const sessionId = 'dc8c9930437f65d30f6597768800957017bac203a0a50342932757c8dfa158d6'; // Use the session ID for symbol lookup

190
apps/data-service/src/providers/webshare.provider.ts
View file

@ -1,5 +1,5 @@
/**
* WebShare Provider for proxy management
* WebShare Provider for proxy management with scheduled updates
*/
import { getLogger } from '@stock-bot/logger';
import {
@ -7,24 +7,10 @@ import {
handlerRegistry,
type HandlerConfigWithSchedule,
} from '@stock-bot/queue';
import { proxyManager } from '@stock-bot/utils';
const logger = getLogger('webshare-provider');
// In-memory proxy storage
let proxies: string[] = [];
let lastFetchTime: Date | null = null;
let currentProxyIndex = 0;
export function getProxy(): string | null {
if (proxies.length === 0) {
return null;
}
const proxy = proxies[currentProxyIndex];
currentProxyIndex = (currentProxyIndex + 1) % proxies.length;
return proxy ?? null;
}
// Initialize and register the WebShare provider
export function initializeWebShareProvider() {
logger.debug('Registering WebShare provider with scheduled jobs...');
@ -34,128 +20,126 @@ export function initializeWebShareProvider() {
operations: {
'fetch-proxies': createJobHandler(async () => {
logger.debug('Fetching proxies from WebShare API');
logger.info('Fetching proxies from WebShare API');
const { fetchWebShareProxies } = await import('./webshare.tasks');
try {
const fetchedProxies = await fetchProxiesFromWebShare();
if (fetchedProxies && fetchedProxies.length > 0) {
proxies = fetchedProxies;
lastFetchTime = new Date();
logger.info('Successfully updated proxy list', {
const proxies = await fetchWebShareProxies();
if (proxies.length > 0) {
// Update the centralized proxy manager
await proxyManager.updateProxies(proxies);
logger.info('Updated proxy manager with WebShare proxies', {
count: proxies.length,
lastFetchTime: lastFetchTime.toISOString(),
workingCount: proxies.filter(p => p.isWorking !== false).length,
});
return {
return {
success: true,
count: proxies.length,
lastFetchTime: lastFetchTime.toISOString(),
proxiesUpdated: proxies.length,
workingProxies: proxies.filter(p => p.isWorking !== false).length,
};
} else {
logger.warn('No proxies fetched from WebShare API');
return {
success: false,
count: 0,
proxiesUpdated: 0,
error: 'No proxies returned from API',
};
}
} catch (error) {
logger.error('Failed to fetch proxies from WebShare', { error });
logger.error('Failed to fetch and update proxies', { error });
return {
success: false,
count: proxies.length,
proxiesUpdated: 0,
error: error instanceof Error ? error.message : 'Unknown error',
};
}
}),
'validate-proxies': createJobHandler(async () => {
logger.info('Validating existing proxies');
const { validateStoredProxies } = await import('./webshare.tasks');
try {
const validationResults = await validateStoredProxies();
// Update proxy manager with validated proxies
await proxyManager.updateProxies(validationResults.workingProxies);
logger.info('Proxy validation completed', {
totalChecked: validationResults.totalChecked,
workingCount: validationResults.workingCount,
successRate: ((validationResults.workingCount / validationResults.totalChecked) * 100).toFixed(1) + '%',
});
return validationResults;
} catch (error) {
logger.error('Failed to validate proxies', { error });
return {
workingProxies: [],
totalChecked: 0,
workingCount: 0,
error: error instanceof Error ? error.message : 'Unknown error',
};
}
}),
'get-stats': createJobHandler(async () => {
const stats = proxyManager.getStats();
logger.info('Proxy manager statistics', stats);
return stats;
}),
},
scheduledJobs: [
{
type: 'fetch-proxies',
type: 'webshare-fetch',
operation: 'fetch-proxies',
payload: {},
description: 'Fetch proxies from WebShare API',
cronPattern: '*/5 * * * *', // Every 5 minutes
priority: 2,
immediately: true, // Fetch immediately on startup
cronPattern: '0 */6 * * *', // Every 6 hours
priority: 3,
description: 'Fetch fresh proxies from WebShare API',
immediately: true, // Run on startup
},
{
type: 'webshare-validate',
operation: 'validate-proxies',
payload: {},
cronPattern: '0 */2 * * *', // Every 2 hours
priority: 4,
description: 'Validate and clean existing proxies',
},
{
type: 'webshare-stats',
operation: 'get-stats',
payload: {},
cronPattern: '0 * * * *', // Every hour
priority: 5,
description: 'Log proxy manager statistics',
},
],
};
// Register the provider
handlerRegistry.registerWithSchedule(webShareProviderConfig);
logger.debug('WebShare provider registered successfully');
}
// Legacy function for backward compatibility - now uses centralized proxy manager
export async function getProxy(): Promise<string | null> {
const proxy = await proxyManager.getRandomProxy();
if (!proxy) {
return null;
}
// Convert ProxyInfo back to string format for backward compatibility
const auth = proxy.username && proxy.password ? `${proxy.username}:${proxy.password}@` : '';
return `${proxy.protocol}://${auth}${proxy.host}:${proxy.port}`;
}
export const webShareProvider = {
initialize: initializeWebShareProvider,
getProxy,
};
async function fetchProxiesFromWebShare(): Promise<string[] | null> {
try {
// Get configuration from config system
const { getConfig } = await import('@stock-bot/config');
const config = getConfig();
// Get configuration from config system
const apiKey = config.webshare?.apiKey;
const apiUrl = config.webshare?.apiUrl;
if (!apiKey || !apiUrl) {
logger.error('Missing WebShare configuration', {
hasApiKey: !!apiKey,
hasApiUrl: !!apiUrl,
configApiKey: apiKey?.substring(0, 10) + '...',
configApiUrl: apiUrl,
});
return null;
}
logger.info('Fetching proxies from WebShare API');
const response = await fetch(`${apiUrl}proxy/list/?mode=direct&page=1&page_size=100`, {
method: 'GET',
headers: {
Authorization: `Token ${apiKey}`,
'Content-Type': 'application/json',
},
});
if (!response.ok) {
logger.error('WebShare API request failed', {
status: response.status,
statusText: response.statusText,
});
return null;
}
const data = await response.json();
if (!data.results || !Array.isArray(data.results)) {
logger.error('Invalid response format from WebShare API', { data });
return null;
}
// Transform proxy data to the format: http://username:password@host:port
const fetchedProxies = data.results.map(
(proxy: { username: string; password: string; proxy_address: string; port: number }) => {
return `http://${proxy.username}:${proxy.password}@${proxy.proxy_address}:${proxy.port}`;
}
);
logger.info('Successfully fetched proxies from WebShare', {
count: fetchedProxies.length,
total: data.count || fetchedProxies.length,
});
// console.log('Fetched Proxies:', fetchedProxies);
return fetchedProxies;
} catch (error) {
logger.error('Failed to fetch proxies from WebShare', { error });
return null;
}
}

253
apps/data-service/src/providers/webshare.tasks.ts Normal file
View file

@ -0,0 +1,253 @@
/**
* WebShare Tasks - API integration and proxy validation
*/
import { getLogger } from '@stock-bot/logger';
import { HttpClient, type ProxyInfo } from '@stock-bot/http';
import { proxyManager } from '@stock-bot/utils';
const logger = getLogger('webshare-tasks');
/**
* Fetch proxies from WebShare API and convert to ProxyInfo format
*/
export async function fetchWebShareProxies(): Promise<ProxyInfo[]> {
try {
// Get configuration from config system
const { getConfig } = await import('@stock-bot/config');
const config = getConfig();
const apiKey = config.webshare?.apiKey;
const apiUrl = config.webshare?.apiUrl;
if (!apiKey || !apiUrl) {
logger.error('Missing WebShare configuration', {
hasApiKey: !!apiKey,
hasApiUrl: !!apiUrl,
});
return [];
}
logger.info('Fetching proxies from WebShare API', { apiUrl });
const response = await fetch(`${apiUrl}proxy/list/?mode=direct&page=1&page_size=100`, {
method: 'GET',
headers: {
Authorization: `Token ${apiKey}`,
'Content-Type': 'application/json',
},
});
if (!response.ok) {
logger.error('WebShare API request failed', {
status: response.status,
statusText: response.statusText,
});
return [];
}
const data = await response.json();
if (!data.results || !Array.isArray(data.results)) {
logger.error('Invalid response format from WebShare API', { data });
return [];
}
// Transform proxy data to ProxyInfo format
const proxies: ProxyInfo[] = data.results.map((proxy: {
username: string;
password: string;
proxy_address: string;
port: number;
}) => ({
source: 'webshare',
protocol: 'http' as const,
host: proxy.proxy_address,
port: proxy.port,
username: proxy.username,
password: proxy.password,
isWorking: true, // Assume working until proven otherwise
firstSeen: new Date(),
lastChecked: new Date(),
}));
logger.info('Successfully fetched proxies from WebShare', {
count: proxies.length,
total: data.count || proxies.length,
});
return proxies;
} catch (error) {
logger.error('Failed to fetch proxies from WebShare', { error });
return [];
}
}
/**
* Validate stored proxies by testing connectivity
*/
export async function validateStoredProxies(): Promise<{
workingProxies: ProxyInfo[];
totalChecked: number;
workingCount: number;
}> {
const httpClient = new HttpClient({ timeout: 10000 }, logger);
const testUrl = 'https://httpbin.org/ip'; // Simple IP echo service
// Get all proxies from proxy manager
const allProxies = await proxyManager.getAllProxies();
if (allProxies.length === 0) {
logger.warn('No proxies available for validation');
return {
workingProxies: [],
totalChecked: 0,
workingCount: 0,
};
}
logger.info('Starting proxy validation', { totalProxies: allProxies.length });
const workingProxies: ProxyInfo[] = [];
const validationPromises = allProxies.map(async (proxy) => {
try {
const startTime = Date.now();
const response = await httpClient.get(testUrl, {
proxy,
timeout: 10000,
});
const responseTime = Date.now() - startTime;
const isWorking = response.status === 200;
if (isWorking) {
// Update proxy with success metrics
const updatedProxy: ProxyInfo = {
...proxy,
isWorking: true,
lastChecked: new Date(),
responseTime,
total: (proxy.total || 0) + 1,
working: (proxy.working || 0) + 1,
averageResponseTime: proxy.averageResponseTime
? (proxy.averageResponseTime + responseTime) / 2
: responseTime,
};
// Calculate success rate
updatedProxy.successRate = updatedProxy.total > 0
? (updatedProxy.working / updatedProxy.total) * 100
: 100;
workingProxies.push(updatedProxy);
logger.debug('Proxy validation successful', {
host: proxy.host,
port: proxy.port,
responseTime,
successRate: updatedProxy.successRate?.toFixed(1) + '%',
});
} else {
logger.debug('Proxy validation failed', {
host: proxy.host,
port: proxy.port,
status: response.status,
});
}
} catch (error) {
// Update proxy with failure metrics
const updatedProxy: ProxyInfo = {
...proxy,
isWorking: false,
lastChecked: new Date(),
error: error instanceof Error ? error.message : 'Unknown error',
total: (proxy.total || 0) + 1,
working: proxy.working || 0,
};
// Calculate success rate
updatedProxy.successRate = updatedProxy.total > 0
? (updatedProxy.working / updatedProxy.total) * 100
: 0;
logger.debug('Proxy validation error', {
host: proxy.host,
port: proxy.port,
error: error instanceof Error ? error.message : 'Unknown error',
});
}
});
// Wait for all validations to complete
await Promise.all(validationPromises);
const results = {
workingProxies,
totalChecked: allProxies.length,
workingCount: workingProxies.length,
};
logger.info('Proxy validation completed', {
totalChecked: results.totalChecked,
workingCount: results.workingCount,
successRate: ((results.workingCount / results.totalChecked) * 100).toFixed(1) + '%',
});
return results;
}
/**
* Test a single proxy for connectivity
*/
export async function testProxy(proxy: ProxyInfo): Promise<ProxyInfo> {
const httpClient = new HttpClient({ timeout: 10000 }, logger);
const testUrl = 'https://httpbin.org/ip';
try {
const startTime = Date.now();
const response = await httpClient.get(testUrl, {
proxy,
timeout: 10000,
});
const responseTime = Date.now() - startTime;
const isWorking = response.status === 200;
const updatedProxy: ProxyInfo = {
...proxy,
isWorking,
lastChecked: new Date(),
responseTime,
total: (proxy.total || 0) + 1,
working: isWorking ? (proxy.working || 0) + 1 : (proxy.working || 0),
};
// Calculate success rate
updatedProxy.successRate = updatedProxy.total > 0
? (updatedProxy.working / updatedProxy.total) * 100
: (isWorking ? 100 : 0);
// Update average response time
if (isWorking && responseTime) {
updatedProxy.averageResponseTime = proxy.averageResponseTime
? (proxy.averageResponseTime + responseTime) / 2
: responseTime;
}
return updatedProxy;
} catch (error) {
const updatedProxy: ProxyInfo = {
...proxy,
isWorking: false,
lastChecked: new Date(),
error: error instanceof Error ? error.message : 'Unknown error',
total: (proxy.total || 0) + 1,
working: proxy.working || 0,
};
updatedProxy.successRate = updatedProxy.total > 0
? (updatedProxy.working / updatedProxy.total) * 100
: 0;
return updatedProxy;
}
}

1
libs/utils/src/index.ts
View file

@ -2,3 +2,4 @@ export * from './calculations/index';
export * from './common';
export * from './dateUtils';
export * from './generic-functions';
export * from './proxy';

5
libs/utils/src/proxy/index.ts Normal file
View file

@ -0,0 +1,5 @@
/**
* Proxy management utilities
*/
export { ProxyManager, proxyManager } from './proxy-manager';
export type { ProxyInfo } from '@stock-bot/http'; // Re-export for convenience

256
libs/utils/src/proxy/proxy-manager.ts Normal file
View file

@ -0,0 +1,256 @@
/**
* Centralized Proxy Manager - Handles proxy storage, retrieval, and caching
*/
import { createCache, type CacheProvider } from '@stock-bot/cache';
import { getDatabaseConfig } from '@stock-bot/config';
import { getLogger } from '@stock-bot/logger';
import type { ProxyInfo } from '@stock-bot/http';
const logger = getLogger('proxy-manager');
export class ProxyManager {
private cache: CacheProvider;
private proxies: ProxyInfo[] = [];
private lastUpdate: Date | null = null;
private isInitialized = false;
constructor() {
const databaseConfig = getDatabaseConfig();
this.cache = createCache({
redisConfig: databaseConfig.dragonfly,
keyPrefix: 'proxies:',
ttl: 86400, // 24 hours
enableMetrics: true,
});
}
/**
* Initialize the proxy manager - loads existing proxies from cache
*/
async initialize(): Promise<void> {
if (this.isInitialized) {
return;
}
try {
logger.info('Initializing proxy manager...');
await this.loadFromCache();
this.isInitialized = true;
logger.info('Proxy manager initialized', {
proxiesLoaded: this.proxies.length,
lastUpdate: this.lastUpdate,
});
} catch (error) {
logger.error('Failed to initialize proxy manager', { error });
this.isInitialized = true; // Set to true anyway to avoid infinite retries
}
}
/**
* Get a random working proxy from the available pool
*/
async getRandomProxy(): Promise<ProxyInfo | null> {
// Ensure initialized
if (!this.isInitialized) {
await this.initialize();
}
// Load from cache if memory is empty
if (this.proxies.length === 0) {
await this.loadFromCache();
}
// Filter for working proxies (not explicitly marked as non-working)
const workingProxies = this.proxies.filter(proxy => proxy.isWorking !== false);
if (workingProxies.length === 0) {
logger.warn('No working proxies available');
return null;
}
// Return random proxy with preference for recently successful ones
const sortedProxies = workingProxies.sort((a, b) => {
// Prefer proxies with better success rates
const aRate = a.successRate || 0;
const bRate = b.successRate || 0;
return bRate - aRate;
});
// Take from top 50% of best performing proxies
const topProxies = sortedProxies.slice(0, Math.max(1, Math.floor(sortedProxies.length * 0.5)));
const selectedProxy = topProxies[Math.floor(Math.random() * topProxies.length)];
if (!selectedProxy) {
logger.warn('No proxy selected from available pool');
return null;
}
logger.debug('Selected proxy', {
host: selectedProxy.host,
port: selectedProxy.port,
successRate: selectedProxy.successRate,
totalAvailable: workingProxies.length,
});
return selectedProxy;
}
/**
* Get all working proxies
*/
async getWorkingProxies(): Promise<ProxyInfo[]> {
if (!this.isInitialized) {
await this.initialize();
}
if (this.proxies.length === 0) {
await this.loadFromCache();
}
return this.proxies.filter(proxy => proxy.isWorking !== false);
}
/**
* Get all proxies (working and non-working)
*/
async getAllProxies(): Promise<ProxyInfo[]> {
if (!this.isInitialized) {
await this.initialize();
}
if (this.proxies.length === 0) {
await this.loadFromCache();
}
return [...this.proxies];
}
/**
* Update the proxy pool with new proxies
*/
async updateProxies(proxies: ProxyInfo[]): Promise<void> {
try {
logger.info('Updating proxy pool', { newCount: proxies.length, existingCount: this.proxies.length });
this.proxies = proxies;
this.lastUpdate = new Date();
// Store to cache
await this.cache.set('active-proxies', proxies);
await this.cache.set('last-update', this.lastUpdate.toISOString());
const workingCount = proxies.filter(p => p.isWorking !== false).length;
logger.info('Proxy pool updated successfully', {
totalProxies: proxies.length,
workingProxies: workingCount,
lastUpdate: this.lastUpdate,
});
} catch (error) {
logger.error('Failed to update proxy pool', { error });
throw error;
}
}
/**
* Add or update a single proxy in the pool
*/
async updateProxy(proxy: ProxyInfo): Promise<void> {
const existingIndex = this.proxies.findIndex(
p => p.host === proxy.host && p.port === proxy.port && p.protocol === proxy.protocol
);
if (existingIndex >= 0) {
this.proxies[existingIndex] = { ...this.proxies[existingIndex], ...proxy };
logger.debug('Updated existing proxy', { host: proxy.host, port: proxy.port });
} else {
this.proxies.push(proxy);
logger.debug('Added new proxy', { host: proxy.host, port: proxy.port });
}
// Update cache
await this.updateProxies(this.proxies);
}
/**
* Remove a proxy from the pool
*/
async removeProxy(host: string, port: number, protocol: string): Promise<void> {
const initialLength = this.proxies.length;
this.proxies = this.proxies.filter(
p => !(p.host === host && p.port === port && p.protocol === protocol)
);
if (this.proxies.length < initialLength) {
await this.updateProxies(this.proxies);
logger.debug('Removed proxy', { host, port, protocol });
}
}
/**
* Get proxy statistics
*/
getStats(): {
totalProxies: number;
workingProxies: number;
lastUpdate: Date | null;
successRate: number;
} {
const workingProxies = this.proxies.filter(p => p.isWorking !== false);
const totalSuccessRate = this.proxies.reduce((sum, p) => sum + (p.successRate || 0), 0);
const avgSuccessRate = this.proxies.length > 0 ? totalSuccessRate / this.proxies.length : 0;
return {
totalProxies: this.proxies.length,
workingProxies: workingProxies.length,
lastUpdate: this.lastUpdate,
successRate: avgSuccessRate,
};
}
/**
* Clear all proxies from memory and cache
*/
async clearProxies(): Promise<void> {
this.proxies = [];
this.lastUpdate = null;
await this.cache.del('active-proxies');
await this.cache.del('last-update');
logger.info('Cleared all proxies');
}
/**
* Check if proxy manager is ready
*/
isReady(): boolean {
return this.isInitialized;
}
/**
* Load proxies from cache storage
*/
private async loadFromCache(): Promise<void> {
try {
const cachedProxies = await this.cache.get<ProxyInfo[]>('active-proxies');
const lastUpdateStr = await this.cache.get<string>('last-update');
if (cachedProxies && Array.isArray(cachedProxies)) {
this.proxies = cachedProxies;
this.lastUpdate = lastUpdateStr ? new Date(lastUpdateStr) : null;
logger.debug('Loaded proxies from cache', {
count: this.proxies.length,
lastUpdate: this.lastUpdate,
});
} else {
logger.debug('No cached proxies found');
}
} catch (error) {
logger.error('Failed to load proxies from cache', { error });
}
}
}
// Singleton instance for easy import
export const proxyManager = new ProxyManager();

6
libs/utils/tsconfig.json
View file

@ -6,6 +6,10 @@
},
"include": ["src/**/*"],
"references": [
{ "path": "../types" }
{ "path": "../types" },
{ "path": "../cache" },
{ "path": "../config" },
{ "path": "../logger" },
{ "path": "../http" }
]
}