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

created qm session check

Browse source
This commit is contained in:
Boki 2025-06-26 22:38:53 -04:00
parent e5f505335c
commit b767689470
4 changed files with 235 additions and 155 deletions
Download patch file Download diff file Expand all files Collapse all files

156
apps/stock/data-ingestion/src/handlers/qm/actions/session.action.ts
View file

@ -3,9 +3,10 @@
*/ */
import type { BaseHandler, ExecutionContext } from '@stock-bot/handlers'; import type { BaseHandler, ExecutionContext } from '@stock-bot/handlers';
import { BunRequestInit, getRandomUserAgent } from '@stock-bot/utils'; import { BunRequestInit } from '@stock-bot/utils';
import { QM_CONFIG, QM_SESSION_IDS, SESSION_CONFIG } from '../shared/config'; import { getQmHeaders, QM_CONFIG, QM_SESSION_IDS, SESSION_CONFIG } from '../shared/config';
import { QMSessionManager } from '../shared/session-manager'; import { QMSessionManager } from '../shared/session-manager';
import { QMSession } from '../shared/types';
/** /**
* Check existing sessions and queue creation jobs for needed sessions * Check existing sessions and queue creation jobs for needed sessions
@ -24,29 +25,22 @@ export async function checkSessions(
const sessionManager = QMSessionManager.getInstance(); const sessionManager = QMSessionManager.getInstance();
// Set cache provider if not already set // Initialize with cache provider and logger
if (this.cache) { sessionManager.initialize(this.cache, this.logger);
sessionManager.setCacheProvider(this.cache);
}
// Load sessions from cache if not initialized // Always load fresh data from cache (don't rely on initialization flag)
if (!sessionManager.getInitialized()) {
await sessionManager.loadFromCache(); await sessionManager.loadFromCache();
sessionManager.setInitialized(true);
}
const cleanedCount = sessionManager.cleanupFailedSessions(); // Cleanup failed sessions (this now handles its own cache sync)
const cleanedCount = await sessionManager.cleanupFailedSessions();
// Sync after cleanup
await sessionManager.syncToCache();
// Check which session IDs need more sessions and queue creation jobs // Check which session IDs need more sessions and queue creation jobs
let queuedCount = 0; let queuedCount = 0;
for (const [sessionType, sessionId] of Object.entries(QM_SESSION_IDS)) { for (const [sessionType, sessionId] of Object.entries(QM_SESSION_IDS)) {
this.logger.debug(`Checking session ID: ${sessionId}`); this.logger.debug(`Checking session ID: ${sessionId}`);
if (sessionManager.needsMoreSessions(sessionId)) { if (await sessionManager.needsMoreSessions(sessionId)) {
const currentCount = sessionManager.getSessions(sessionId).length; const currentCount = sessionManager.getSessions(sessionId).length;
const neededSessions = SESSION_CONFIG.MIN_SESSIONS - currentCount; const neededSessions = SESSION_CONFIG.MAX_SESSIONS - currentCount;
// Queue up to 10 at a time to avoid overwhelming the system // Queue up to 10 at a time to avoid overwhelming the system
const toQueue = Math.min(neededSessions, 10); const toQueue = Math.min(neededSessions, 10);
@ -60,7 +54,7 @@ export async function checkSessions(
this.logger.info(`Queued ${toQueue} jobs to create sessions for ${sessionType}`, { this.logger.info(`Queued ${toQueue} jobs to create sessions for ${sessionType}`, {
currentCount, currentCount,
targetCount: SESSION_CONFIG.MIN_SESSIONS, targetCount: SESSION_CONFIG.MAX_SESSIONS,
}); });
} }
} }
@ -88,7 +82,7 @@ interface CreateSessionInput {
export async function createSession( export async function createSession(
this: BaseHandler, this: BaseHandler,
input: CreateSessionInput input: CreateSessionInput
): Promise<{ sessionId: string; status: string; sessionType: string }> { ): Promise<{ sessionId: string; status: string; sessionType: string, session?: QMSession }> {
const { sessionId, sessionType = 'LOOKUP' } = input || {}; const { sessionId, sessionType = 'LOOKUP' } = input || {};
const sessionManager = QMSessionManager.getInstance(); const sessionManager = QMSessionManager.getInstance();
@ -99,10 +93,8 @@ export async function createSession(
throw new Error(`Invalid session type: ${sessionType}`); throw new Error(`Invalid session type: ${sessionType}`);
} }
// Set cache provider if not already set // Initialize with cache provider and logger
if (this.cache) { sessionManager.initialize(this.cache, this.logger);
sessionManager.setCacheProvider(this.cache);
}
try { try {
// Get proxy from proxy service // Get proxy from proxy service
@ -112,95 +104,59 @@ export async function createSession(
throw new Error(`No proxy available for session type ${sessionType}`); throw new Error(`No proxy available for session type ${sessionType}`);
} }
const userAgent = getRandomUserAgent();
this.logger.debug(`Using User-Agent: ${userAgent}, proxy: ${proxyUrl || 'none'}`);
// Authenticate with QM API inline // Authenticate with QM API inline
const authUrl = `${QM_CONFIG.BASE_URL}${QM_CONFIG.SESSION_PATH}`; const sessionUrl = `${QM_CONFIG.BASE_URL}${QM_CONFIG.SESSION_PATH}/${sessionId}`;
// Build request options // Build request options
const requestOptions: BunRequestInit = { const sessionRequest: BunRequestInit = {
method: 'GET',
proxy: proxyUrl || undefined, proxy: proxyUrl || undefined,
headers: { headers: getQmHeaders(),
'User-Agent': userAgent,
Accept: '*/*',
'Accept-Language': 'en',
'Sec-Fetch-Mode': 'cors',
Origin: 'https://www.quotemedia.com',
Referer: 'https://www.quotemedia.com/',
},
redirect: 'manual', // Don't follow redirects automatically
}; };
this.logger.debug('Authenticating with QM API', { authUrl }); this.logger.debug('Authenticating with QM API', { sessionUrl, sessionRequest });
const response = await fetch(authUrl, requestOptions); const sessionResponse = await fetch(sessionUrl, sessionRequest);
// Extract cookies from response headers // Check if authentication was successful
// const cookies: string[] = []; if (sessionResponse.status === 200 || sessionResponse.status === 302) {
// const setCookieHeaders = response.headers.getSetCookie(); this.logger.info('QM authentication successful', {
status: sessionResponse.status,
});
}else{
this.logger.warn('QM authentication failed', {
status: sessionResponse.status,
statusText: sessionResponse.statusText,
});
throw new Error(`QM authentication failed with status ${sessionResponse.status}`);
}
// if (setCookieHeaders && setCookieHeaders.length > 0) {
// cookies.push(...setCookieHeaders);
// }
// // Check if authentication was successful const sessionData = await sessionResponse.json();
// if (response.status === 200 || response.status === 302) {
// this.logger.info('QM authentication successful', {
// status: response.status,
// cookieCount: cookies.length,
// });
// // Build headers with cookies // Add token to headers
// const headers = sessionManager.getQmHeaders(); sessionRequest.headers['Datatool-Token'] = sessionData.token;
// if (cookies.length > 0) {
// headers['Cookie'] = buildCookieString(cookies);
// }
// // Create session object // Create session object with unique ID
// const session: QMSession = { const session: QMSession = {
// proxy: proxyUrl || '', uuid: `${sessionType.toLowerCase()}_${Date.now()}_${Math.random().toString(36).substring(2, 9)}`,
// headers, proxy: proxyUrl,
// successfulCalls: 0, headers: sessionRequest.headers,
// failedCalls: 0, successfulCalls: 0,
// lastUsed: new Date(), failedCalls: 0,
// }; lastUsed: new Date(),
createdAt: new Date(),
};
// // Add session to manager // Add session to manager (this now handles cache sync)
// sessionManager.addSession(actualSessionId, session); await sessionManager.addSession(actualSessionId, session);
// // Sync to cache this.logger.info(`Successfully created session for ${sessionType}`, { session });
// await sessionManager.syncToCache();
// this.logger.info(`Successfully created session for ${sessionType}`, {
// sessionId: actualSessionId,
// hasProxy: !!proxyUrl,
// hasCookies: cookies.length > 0,
// });
// return {
// sessionId: actualSessionId,
// status: 'created',
// sessionType,
// };
// } else {
// this.logger.warn('QM authentication failed', {
// status: response.status,
// statusText: response.statusText,
// });
// return {
// sessionId: actualSessionId,
// status: 'failed',
// sessionType,
// };
// }
return { return {
sessionId: 'test',//actualSessionId, sessionId: actualSessionId,
status: 'created', status: 'created',
sessionType, sessionType,
session,
}; };
} catch (error) { } catch (error) {
this.logger.error(`Failed to create session for ${sessionType}`, { error }); this.logger.error(`Failed to create session for ${sessionType}`, { error });
@ -211,17 +167,3 @@ export async function createSession(
}; };
} }
} }
/**
* Build cookie string from array of set-cookie headers
*/
function buildCookieString(cookies: string[]): string {
return cookies
.map(cookie => {
// Extract just the name=value part, ignore attributes
const match = cookie.match(/^([^=]+=[^;]+)/);
return match ? match[1] : '';
})
.filter(Boolean)
.join('; ');
}

17
apps/stock/data-ingestion/src/handlers/qm/shared/config.ts
View file

@ -2,6 +2,8 @@
* Shared configuration for QM operations * Shared configuration for QM operations
*/ */
import { getRandomUserAgent } from "@stock-bot/utils";
// QM Session IDs for different endpoints // QM Session IDs for different endpoints
export const QM_SESSION_IDS = { export const QM_SESSION_IDS = {
LOOKUP: 'dc8c9930437f65d30f6597768800957017bac203a0a50342932757c8dfa158d6', // lookup endpoint LOOKUP: 'dc8c9930437f65d30f6597768800957017bac203a0a50342932757c8dfa158d6', // lookup endpoint
@ -30,9 +32,20 @@ export const QM_CONFIG = {
// Session management settings // Session management settings
export const SESSION_CONFIG = { export const SESSION_CONFIG = {
MIN_SESSIONS: 15, MIN_SESSIONS: 2,
MAX_SESSIONS: 50, MAX_SESSIONS: 5,
MAX_FAILED_CALLS: 3, MAX_FAILED_CALLS: 3,
SESSION_TIMEOUT: 5000, // 10 seconds SESSION_TIMEOUT: 5000, // 10 seconds
API_TIMEOUT: 30000, // 15 seconds API_TIMEOUT: 30000, // 15 seconds
} as const; } as const;
export function getQmHeaders(): Record<string, string> {
return {
'User-Agent': getRandomUserAgent(),
Accept: '*/*',
'Accept-Language': 'en',
'Sec-Fetch-Mode': 'cors',
Origin: 'https://www.quotemedia.com',
Referer: 'https://www.quotemedia.com/',
};
}

185
apps/stock/data-ingestion/src/handlers/qm/shared/session-manager.ts
View file

@ -3,7 +3,7 @@
*/ */
import type { CacheProvider } from '@stock-bot/cache'; import type { CacheProvider } from '@stock-bot/cache';
import { getRandomUserAgent } from '@stock-bot/utils'; import type { Logger } from '@stock-bot/types';
import { QM_SESSION_IDS, SESSION_CONFIG } from './config'; import { QM_SESSION_IDS, SESSION_CONFIG } from './config';
import type { CachedSession, QMSession } from './types'; import type { CachedSession, QMSession } from './types';
@ -12,6 +12,7 @@ export class QMSessionManager {
private sessionCache: Record<string, QMSession[]> = {}; private sessionCache: Record<string, QMSession[]> = {};
private isInitialized = false; private isInitialized = false;
private cacheProvider: CacheProvider | null = null; private cacheProvider: CacheProvider | null = null;
private logger: Logger | null = null;
private constructor() { private constructor() {
// Initialize session cache with known session IDs // Initialize session cache with known session IDs
@ -27,6 +28,16 @@ export class QMSessionManager {
return QMSessionManager.instance; return QMSessionManager.instance;
} }
/**
* Reset the singleton instance (for testing only)
*/
static resetInstance(): void {
if (QMSessionManager.instance?.logger) {
QMSessionManager.instance.logger.warn('Resetting QMSessionManager instance - this should only be used for testing');
}
QMSessionManager.instance = null;
}
/** /**
* Set the cache provider for persistence * Set the cache provider for persistence
*/ */
@ -34,10 +45,33 @@ export class QMSessionManager {
this.cacheProvider = cache; this.cacheProvider = cache;
} }
/**
* Set the logger
*/
setLogger(logger: Logger): void {
this.logger = logger;
this.logger.trace('Logger set for QMSessionManager');
}
/**
* Initialize with cache provider and logger
*/
initialize(cache?: CacheProvider, logger?: Logger): void {
if (cache) {
this.setCacheProvider(cache);
}
if (logger) {
this.setLogger(logger);
}
}
/** /**
* Get a random session for the given session ID * Get a random session for the given session ID
*/ */
getSession(sessionId: string): QMSession | null { async getSession(sessionId: string): Promise<QMSession | null> {
// Always load fresh data from cache
await this.loadFromCache();
const sessions = this.sessionCache[sessionId]; const sessions = this.sessionCache[sessionId];
if (!sessions || sessions.length === 0) { if (!sessions || sessions.length === 0) {
return null; return null;
@ -54,14 +88,33 @@ export class QMSessionManager {
return validSessions[Math.floor(Math.random() * validSessions.length)]; return validSessions[Math.floor(Math.random() * validSessions.length)];
} }
/**
* Get a specific session by UUID
*/
getSessionByUuid(sessionId: string, uuid: string): QMSession | null {
const sessions = this.sessionCache[sessionId] || [];
return sessions.find(s => s.uuid === uuid) || null;
}
/** /**
* Add a session to the cache * Add a session to the cache
*/ */
addSession(sessionId: string, session: QMSession): void { async addSession(sessionId: string, session: QMSession): Promise<void> {
// Load latest from cache first to avoid overwriting other service's changes
await this.loadFromCache();
if (!this.sessionCache[sessionId]) { if (!this.sessionCache[sessionId]) {
this.sessionCache[sessionId] = []; this.sessionCache[sessionId] = [];
} }
this.sessionCache[sessionId].push(session); this.sessionCache[sessionId].push(session);
this.logger?.debug(`Added session ${session.uuid} to ${sessionId}`, {
sessionId,
uuid: session.uuid
});
// Sync to cache immediately
await this.syncToCache();
} }
/** /**
@ -81,7 +134,10 @@ export class QMSessionManager {
/** /**
* Clean up failed sessions * Clean up failed sessions
*/ */
cleanupFailedSessions(): number { async cleanupFailedSessions(): Promise<number> {
// Always load latest from cache first
await this.loadFromCache();
let removedCount = 0; let removedCount = 0;
Object.keys(this.sessionCache).forEach(sessionId => { Object.keys(this.sessionCache).forEach(sessionId => {
@ -92,29 +148,24 @@ export class QMSessionManager {
removedCount += initialCount - this.sessionCache[sessionId].length; removedCount += initialCount - this.sessionCache[sessionId].length;
}); });
return removedCount; // Sync back to cache after cleanup
} await this.syncToCache();
getQmHeaders(): Record<string, string> { return removedCount;
return {
'User-Agent': getRandomUserAgent(),
Accept: '*/*',
'Accept-Language': 'en',
'Sec-Fetch-Mode': 'cors',
Origin: 'https://www.quotemedia.com',
Referer: 'https://www.quotemedia.com/',
};
} }
/** /**
* Check if more sessions are needed for a session ID * Check if more sessions are needed for a session ID
*/ */
needsMoreSessions(sessionId: string): boolean { async needsMoreSessions(sessionId: string): Promise<boolean> {
// Always load fresh data from cache
await this.loadFromCache();
const sessions = this.sessionCache[sessionId] || []; const sessions = this.sessionCache[sessionId] || [];
const validSessions = sessions.filter( const validSessions = sessions.filter(
session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS
); );
return validSessions.length < SESSION_CONFIG.MIN_SESSIONS; return validSessions.length < SESSION_CONFIG.MAX_SESSIONS;
} }
/** /**
@ -150,14 +201,14 @@ export class QMSessionManager {
} }
/** /**
* Mark manager as initialized * Mark manager as initialized (deprecated - we always load from cache now)
*/ */
setInitialized(initialized: boolean = true): void { setInitialized(initialized: boolean = true): void {
this.isInitialized = initialized; this.isInitialized = initialized;
} }
/** /**
* Check if manager is initialized * Check if manager is initialized (deprecated - we always load from cache now)
*/ */
getInitialized(): boolean { getInitialized(): boolean {
return this.isInitialized; return this.isInitialized;
@ -168,15 +219,20 @@ export class QMSessionManager {
*/ */
async loadFromCache(): Promise<void> { async loadFromCache(): Promise<void> {
if (!this.cacheProvider) { if (!this.cacheProvider) {
this.logger?.warn('No cache provider available for loading sessions');
return; return;
} }
try { try {
this.logger?.trace('Loading sessions from cache...');
// Load sessions for each session type // Load sessions for each session type
for (const [sessionType, sessionId] of Object.entries(QM_SESSION_IDS)) { for (const [sessionType, sessionId] of Object.entries(QM_SESSION_IDS)) {
const listKey = `qm:sessions:${sessionType.toLowerCase()}:list`; const listKey = `qm:sessions:${sessionType.toLowerCase()}:list`;
const sessionIds = await this.cacheProvider.get<string[]>(listKey); const sessionIds = await this.cacheProvider.get<string[]>(listKey);
this.logger?.trace(`Loading ${sessionType} sessions`, { sessionIds });
if (sessionIds && Array.isArray(sessionIds)) { if (sessionIds && Array.isArray(sessionIds)) {
const sessions: QMSession[] = []; const sessions: QMSession[] = [];
@ -185,21 +241,34 @@ export class QMSessionManager {
const cachedSession = await this.cacheProvider.get<CachedSession>(sessionKey); const cachedSession = await this.cacheProvider.get<CachedSession>(sessionKey);
if (cachedSession) { if (cachedSession) {
sessions.push({ const session = {
uuid: cachedSession.uuid,
proxy: cachedSession.proxy, proxy: cachedSession.proxy,
headers: cachedSession.headers, headers: cachedSession.headers,
successfulCalls: cachedSession.successfulCalls, successfulCalls: cachedSession.successfulCalls || 0,
failedCalls: cachedSession.failedCalls, failedCalls: cachedSession.failedCalls || 0,
lastUsed: new Date(cachedSession.lastUsed), lastUsed: new Date(cachedSession.lastUsed),
createdAt: cachedSession.createdAt ? new Date(cachedSession.createdAt) : new Date(),
};
this.logger?.trace(`Loaded session ${id}`, {
uuid: session.uuid,
successfulCalls: session.successfulCalls,
failedCalls: session.failedCalls
}); });
sessions.push(session);
} }
} }
this.sessionCache[sessionId] = sessions; this.sessionCache[sessionId] = sessions;
this.logger?.debug(`Loaded ${sessions.length} sessions for ${sessionType}`);
} else {
this.logger?.trace(`No sessions found for ${sessionType}`);
} }
} }
} catch (error) { } catch (error) {
console.error('Failed to load sessions from cache:', error); this.logger?.error('Failed to load sessions from cache', { error });
} }
} }
@ -216,54 +285,108 @@ export class QMSessionManager {
const sessions = this.sessionCache[sessionId] || []; const sessions = this.sessionCache[sessionId] || [];
const sessionIds: string[] = []; const sessionIds: string[] = [];
// Store each session // Clear old sessions first
const oldListKey = `qm:sessions:${sessionType.toLowerCase()}:list`;
const oldSessionIds = await this.cacheProvider.get<string[]>(oldListKey);
if (oldSessionIds && Array.isArray(oldSessionIds)) {
// Delete old session entries
for (const oldId of oldSessionIds) {
const oldSessionKey = `qm:sessions:${sessionType.toLowerCase()}:${oldId}`;
await this.cacheProvider.del(oldSessionKey);
}
}
// Store each session with stable IDs
for (let i = 0; i < sessions.length; i++) { for (let i = 0; i < sessions.length; i++) {
const session = sessions[i]; const session = sessions[i];
const id = `${sessionType.toLowerCase()}_${i}_${Date.now()}`; const id = `${sessionType.toLowerCase()}_${i}`;
const sessionKey = `qm:sessions:${sessionType.toLowerCase()}:${id}`; const sessionKey = `qm:sessions:${sessionType.toLowerCase()}:${id}`;
const cachedSession: CachedSession = { const cachedSession: CachedSession = {
...session, ...session,
lastUsed: session.lastUsed instanceof Date ? session.lastUsed.toISOString() : session.lastUsed,
createdAt: session.createdAt instanceof Date ? session.createdAt.toISOString() : session.createdAt,
id, id,
sessionType, sessionType,
}; } as any;
this.logger?.trace(`Saving session ${id}`, {
uuid: cachedSession.uuid,
successfulCalls: cachedSession.successfulCalls,
failedCalls: cachedSession.failedCalls
});
await this.cacheProvider.set(sessionKey, cachedSession, 86400); // 24 hour TTL await this.cacheProvider.set(sessionKey, cachedSession, 86400); // 24 hour TTL
sessionIds.push(id); sessionIds.push(id);
} }
// Store the list of session IDs // Store the list of session IDs
const listKey = `qm:sessions:${sessionType.toLowerCase()}:list`; await this.cacheProvider.set(oldListKey, sessionIds, 86400);
await this.cacheProvider.set(listKey, sessionIds, 86400);
} }
// Store stats // Store stats
const statsKey = 'qm:sessions:stats'; const statsKey = 'qm:sessions:stats';
await this.cacheProvider.set(statsKey, this.getStats(), 3600); await this.cacheProvider.set(statsKey, this.getStats(), 3600);
this.logger?.trace('Session sync to cache completed');
} catch (error) { } catch (error) {
console.error('Failed to sync sessions to cache:', error); this.logger?.error('Failed to sync sessions to cache', { error });
} }
} }
/** /**
* Increment failed calls for a session * Increment failed calls for a session
*/ */
async incrementFailedCalls(sessionId: string, session: QMSession): Promise<void> { async incrementFailedCalls(sessionId: string, sessionUuid: string): Promise<void> {
// Load latest from cache first
await this.loadFromCache();
// Find session by UUID
const sessions = this.sessionCache[sessionId] || [];
const session = sessions.find(s => s.uuid === sessionUuid);
if (session) {
session.failedCalls++; session.failedCalls++;
session.lastUsed = new Date(); session.lastUsed = new Date();
this.logger?.debug(`Incremented failed calls for session`, {
sessionUuid,
failedCalls: session.failedCalls,
sessionId
});
// Sync to cache after update // Sync to cache after update
await this.syncToCache(); await this.syncToCache();
} else {
this.logger?.warn(`Session not found`, { sessionUuid, sessionId });
}
} }
/** /**
* Increment successful calls for a session * Increment successful calls for a session
*/ */
async incrementSuccessfulCalls(sessionId: string, session: QMSession): Promise<void> { async incrementSuccessfulCalls(sessionId: string, sessionUuid: string): Promise<void> {
// Load latest from cache first
await this.loadFromCache();
// Find session by UUID
const sessions = this.sessionCache[sessionId] || [];
const session = sessions.find(s => s.uuid === sessionUuid);
if (session) {
session.successfulCalls++; session.successfulCalls++;
session.lastUsed = new Date(); session.lastUsed = new Date();
this.logger?.debug(`Incremented successful calls for session`, {
sessionUuid,
successfulCalls: session.successfulCalls,
sessionId
});
// Sync to cache after update // Sync to cache after update
await this.syncToCache(); await this.syncToCache();
} else {
this.logger?.warn(`Session not found`, { sessionUuid, sessionId });
}
} }
} }

4
apps/stock/data-ingestion/src/handlers/qm/shared/types.ts
View file

@ -3,11 +3,13 @@
*/ */
export interface QMSession { export interface QMSession {
uuid?: string; // Unique identifier for the session
proxy: string; proxy: string;
headers: Record<string, string>; headers: HeadersInit;
successfulCalls: number; successfulCalls: number;
failedCalls: number; failedCalls: number;
lastUsed: Date; lastUsed: Date;
createdAt: Date;
} }
export interface SymbolSpiderJob { export interface SymbolSpiderJob {