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

format

Browse source
This commit is contained in:
Boki 2025-06-22 17:55:51 -04:00
parent d858222af7
commit 7d9044ab29
202 changed files with 10755 additions and 10972 deletions
Download patch file Download diff file Expand all files Collapse all files

2
apps/data-ingestion/config/default.json
View file

@ -94,4 +94,4 @@
"burstSize": 20
}
}
}
}

6
apps/data-ingestion/src/handlers/ceo/actions/index.ts
View file

@ -1,3 +1,3 @@
export { updateCeoChannels } from './update-ceo-channels.action';
export { updateUniqueSymbols } from './update-unique-symbols.action';
export { processIndividualSymbol } from './process-individual-symbol.action';
export { updateCeoChannels } from './update-ceo-channels.action';
export { updateUniqueSymbols } from './update-unique-symbols.action';
export { processIndividualSymbol } from './process-individual-symbol.action';

228
apps/data-ingestion/src/handlers/ceo/actions/process-individual-symbol.action.ts
View file

@ -1,111 +1,117 @@
import { getRandomUserAgent } from '@stock-bot/utils';
import type { CeoHandler } from '../ceo.handler';
export async function processIndividualSymbol(this: CeoHandler, payload: any, _context: any): Promise<unknown> {
const { ceoId, symbol, timestamp } = payload;
const proxy = this.proxy?.getProxy();
if(!proxy) {
this.logger.warn('No proxy available for processing individual CEO symbol');
return;
}
this.logger.debug('Processing individual CEO symbol', {
ceoId,
timestamp,
});
try {
// Fetch detailed information for the individual symbol
const response = await this.http.get(`https://api.ceo.ca/api/get_spiels?channel=${ceoId}&load_more=top`
+ (timestamp ? `&until=${timestamp}` : ''),
{
proxy: proxy,
headers: {
'User-Agent': getRandomUserAgent()
}
});
if (!response.ok) {
throw new Error(`Failed to fetch details for ceoId ${ceoId}: ${response.statusText}`);
}
const data = await response.json();
const spielCount = data.spiels.length;
if(spielCount === 0) {
this.logger.warn(`No spiels found for ceoId ${ceoId}`);
return null; // No data to process
}
const latestSpielTime = data.spiels[0]?.timestamp;
const posts = data.spiels.map((spiel: any) => ({
ceoId,
spiel: spiel.spiel,
spielReplyToId: spiel.spiel_reply_to_id,
spielReplyTo: spiel.spiel_reply_to,
spielReplyToName: spiel.spiel_reply_to_name,
spielReplyToEdited: spiel.spiel_reply_to_edited,
userId: spiel.user_id,
name: spiel.name,
timestamp: spiel.timestamp,
spielId: spiel.spiel_id,
color: spiel.color,
parentId: spiel.parent_id,
publicId: spiel.public_id,
parentChannel: spiel.parent_channel,
parentTimestamp: spiel.parent_timestamp,
votes: spiel.votes,
editable: spiel.editable,
edited: spiel.edited,
featured: spiel.featured,
verified: spiel.verified,
fake: spiel.fake,
bot: spiel.bot,
voted: spiel.voted,
flagged: spiel.flagged,
ownSpiel: spiel.own_spiel,
score: spiel.score,
savedId: spiel.saved_id,
savedTimestamp: spiel.saved_timestamp,
poll: spiel.poll,
votedInPoll: spiel.voted_in_poll
}));
await this.mongodb.batchUpsert('ceoPosts', posts, ['spielId']);
this.logger.info(`Fetched ${spielCount} spiels for ceoId ${ceoId}`);
// Update Shorts
const shortRes = await this.http.get(`https://api.ceo.ca/api/short_positions/one?symbol=${symbol}`,
{
proxy: proxy,
headers: {
'User-Agent': getRandomUserAgent()
}
});
if (shortRes.ok) {
const shortData = await shortRes.json();
if(shortData && shortData.positions) {
await this.mongodb.batchUpsert('ceoShorts', shortData.positions, ['id']);
}
await this.scheduleOperation('process-individual-symbol', {
ceoId: ceoId,
timestamp: latestSpielTime
});
}
this.logger.info(`Successfully processed channel ${ceoId} and added channel ${ceoId} at timestamp ${latestSpielTime}`);
return { ceoId, spielCount, timestamp };
} catch (error) {
this.logger.error('Failed to process individual symbol', {
error,
ceoId,
timestamp
});
throw error;
}
}
import { getRandomUserAgent } from '@stock-bot/utils';
import type { CeoHandler } from '../ceo.handler';
export async function processIndividualSymbol(
this: CeoHandler,
payload: any,
_context: any
): Promise<unknown> {
const { ceoId, symbol, timestamp } = payload;
const proxy = this.proxy?.getProxy();
if (!proxy) {
this.logger.warn('No proxy available for processing individual CEO symbol');
return;
}
this.logger.debug('Processing individual CEO symbol', {
ceoId,
timestamp,
});
try {
// Fetch detailed information for the individual symbol
const response = await this.http.get(
`https://api.ceo.ca/api/get_spiels?channel=${ceoId}&load_more=top` +
(timestamp ? `&until=${timestamp}` : ''),
{
proxy: proxy,
headers: {
'User-Agent': getRandomUserAgent(),
},
}
);
if (!response.ok) {
throw new Error(`Failed to fetch details for ceoId ${ceoId}: ${response.statusText}`);
}
const data = await response.json();
const spielCount = data.spiels.length;
if (spielCount === 0) {
this.logger.warn(`No spiels found for ceoId ${ceoId}`);
return null; // No data to process
}
const latestSpielTime = data.spiels[0]?.timestamp;
const posts = data.spiels.map((spiel: any) => ({
ceoId,
spiel: spiel.spiel,
spielReplyToId: spiel.spiel_reply_to_id,
spielReplyTo: spiel.spiel_reply_to,
spielReplyToName: spiel.spiel_reply_to_name,
spielReplyToEdited: spiel.spiel_reply_to_edited,
userId: spiel.user_id,
name: spiel.name,
timestamp: spiel.timestamp,
spielId: spiel.spiel_id,
color: spiel.color,
parentId: spiel.parent_id,
publicId: spiel.public_id,
parentChannel: spiel.parent_channel,
parentTimestamp: spiel.parent_timestamp,
votes: spiel.votes,
editable: spiel.editable,
edited: spiel.edited,
featured: spiel.featured,
verified: spiel.verified,
fake: spiel.fake,
bot: spiel.bot,
voted: spiel.voted,
flagged: spiel.flagged,
ownSpiel: spiel.own_spiel,
score: spiel.score,
savedId: spiel.saved_id,
savedTimestamp: spiel.saved_timestamp,
poll: spiel.poll,
votedInPoll: spiel.voted_in_poll,
}));
await this.mongodb.batchUpsert('ceoPosts', posts, ['spielId']);
this.logger.info(`Fetched ${spielCount} spiels for ceoId ${ceoId}`);
// Update Shorts
const shortRes = await this.http.get(
`https://api.ceo.ca/api/short_positions/one?symbol=${symbol}`,
{
proxy: proxy,
headers: {
'User-Agent': getRandomUserAgent(),
},
}
);
if (shortRes.ok) {
const shortData = await shortRes.json();
if (shortData && shortData.positions) {
await this.mongodb.batchUpsert('ceoShorts', shortData.positions, ['id']);
}
await this.scheduleOperation('process-individual-symbol', {
ceoId: ceoId,
timestamp: latestSpielTime,
});
}
this.logger.info(
`Successfully processed channel ${ceoId} and added channel ${ceoId} at timestamp ${latestSpielTime}`
);
return { ceoId, spielCount, timestamp };
} catch (error) {
this.logger.error('Failed to process individual symbol', {
error,
ceoId,
timestamp,
});
throw error;
}
}

139
apps/data-ingestion/src/handlers/ceo/actions/update-ceo-channels.action.ts
View file

@ -1,67 +1,72 @@
import { getRandomUserAgent } from '@stock-bot/utils';
import type { CeoHandler } from '../ceo.handler';
export async function updateCeoChannels(this: CeoHandler, payload: number | undefined): Promise<unknown> {
const proxy = this.proxy?.getProxy();
if(!proxy) {
this.logger.warn('No proxy available for CEO channels update');
return;
}
let page;
if(payload === undefined) {
page = 1
}else{
page = payload;
}
this.logger.info(`Fetching CEO channels for page ${page} with proxy ${proxy}`);
const response = await this.http.get('https://api.ceo.ca/api/home?exchange=all&sort_by=symbol&sector=All&tab=companies&page='+page, {
proxy: proxy,
headers: {
'User-Agent': getRandomUserAgent()
}
})
const results = await response.json();
const channels = results.channel_categories[0].channels;
const totalChannels = results.channel_categories[0].total_channels;
const totalPages = Math.ceil(totalChannels / channels.length);
const exchanges: {exchange: string, countryCode: string}[] = []
const symbols = channels.map((channel: any) =>{
// check if exchange is in the exchanges array object
if(!exchanges.find((e: any) => e.exchange === channel.exchange)) {
exchanges.push({
exchange: channel.exchange,
countryCode: 'CA'
});
}
const details = channel.company_details || {};
return {
symbol: channel.symbol,
exchange: channel.exchange,
name: channel.title,
type: channel.type,
ceoId: channel.channel,
marketCap: details.market_cap,
volumeRatio: details.volume_ratio,
avgVolume: details.avg_volume,
stockType: details.stock_type,
issueType: details.issue_type,
sharesOutstanding: details.shares_outstanding,
float: details.float,
}
})
await this.mongodb.batchUpsert('ceoSymbols', symbols, ['symbol', 'exchange']);
await this.mongodb.batchUpsert('ceoExchanges', exchanges, ['exchange']);
if(page === 1) {
for( let i = 2; i <= totalPages; i++) {
this.logger.info(`Scheduling page ${i} of ${totalPages} for CEO channels`);
await this.scheduleOperation('update-ceo-channels', i)
}
}
this.logger.info(`Fetched CEO channels for page ${page}/${totalPages}`);
return { page, totalPages };
}
import { getRandomUserAgent } from '@stock-bot/utils';
import type { CeoHandler } from '../ceo.handler';
export async function updateCeoChannels(
this: CeoHandler,
payload: number | undefined
): Promise<unknown> {
const proxy = this.proxy?.getProxy();
if (!proxy) {
this.logger.warn('No proxy available for CEO channels update');
return;
}
let page;
if (payload === undefined) {
page = 1;
} else {
page = payload;
}
this.logger.info(`Fetching CEO channels for page ${page} with proxy ${proxy}`);
const response = await this.http.get(
'https://api.ceo.ca/api/home?exchange=all&sort_by=symbol&sector=All&tab=companies&page=' + page,
{
proxy: proxy,
headers: {
'User-Agent': getRandomUserAgent(),
},
}
);
const results = await response.json();
const channels = results.channel_categories[0].channels;
const totalChannels = results.channel_categories[0].total_channels;
const totalPages = Math.ceil(totalChannels / channels.length);
const exchanges: { exchange: string; countryCode: string }[] = [];
const symbols = channels.map((channel: any) => {
// check if exchange is in the exchanges array object
if (!exchanges.find((e: any) => e.exchange === channel.exchange)) {
exchanges.push({
exchange: channel.exchange,
countryCode: 'CA',
});
}
const details = channel.company_details || {};
return {
symbol: channel.symbol,
exchange: channel.exchange,
name: channel.title,
type: channel.type,
ceoId: channel.channel,
marketCap: details.market_cap,
volumeRatio: details.volume_ratio,
avgVolume: details.avg_volume,
stockType: details.stock_type,
issueType: details.issue_type,
sharesOutstanding: details.shares_outstanding,
float: details.float,
};
});
await this.mongodb.batchUpsert('ceoSymbols', symbols, ['symbol', 'exchange']);
await this.mongodb.batchUpsert('ceoExchanges', exchanges, ['exchange']);
if (page === 1) {
for (let i = 2; i <= totalPages; i++) {
this.logger.info(`Scheduling page ${i} of ${totalPages} for CEO channels`);
await this.scheduleOperation('update-ceo-channels', i);
}
}
this.logger.info(`Fetched CEO channels for page ${page}/${totalPages}`);
return { page, totalPages };
}

134
apps/data-ingestion/src/handlers/ceo/actions/update-unique-symbols.action.ts
View file

@ -1,63 +1,71 @@
import type { CeoHandler } from '../ceo.handler';
export async function updateUniqueSymbols(this: CeoHandler, _payload: unknown, _context: any): Promise<unknown> {
this.logger.info('Starting update to get unique CEO symbols by ceoId');
try {
// Get unique ceoId values from the ceoSymbols collection
const uniqueCeoIds = await this.mongodb.collection('ceoSymbols').distinct('ceoId');
this.logger.info(`Found ${uniqueCeoIds.length} unique CEO IDs`);
// Get detailed records for each unique ceoId (latest/first record)
const uniqueSymbols = [];
for (const ceoId of uniqueCeoIds) {
const symbol = await this.mongodb.collection('ceoSymbols')
.findOne({ ceoId }, { sort: { _id: -1 } }); // Get latest record
if (symbol) {
uniqueSymbols.push(symbol);
}
}
this.logger.info(`Retrieved ${uniqueSymbols.length} unique symbol records`);
// Schedule individual jobs for each unique symbol
let scheduledJobs = 0;
for (const symbol of uniqueSymbols) {
// Schedule a job to process this individual symbol
await this.scheduleOperation('process-individual-symbol', {
ceoId: symbol.ceoId,
symbol: symbol.symbol,
});
scheduledJobs++;
// Add small delay to avoid overwhelming the queue
if (scheduledJobs % 10 === 0) {
this.logger.debug(`Scheduled ${scheduledJobs} jobs so far`);
}
}
this.logger.info(`Successfully scheduled ${scheduledJobs} individual symbol update jobs`);
// Cache the results for monitoring
await this.cacheSet('unique-symbols-last-run', {
timestamp: new Date().toISOString(),
totalUniqueIds: uniqueCeoIds.length,
totalRecords: uniqueSymbols.length,
scheduledJobs
}, 1800); // Cache for 30 minutes
return {
success: true,
uniqueCeoIds: uniqueCeoIds.length,
uniqueRecords: uniqueSymbols.length,
scheduledJobs,
timestamp: new Date().toISOString()
};
} catch (error) {
this.logger.error('Failed to update unique CEO symbols', { error });
throw error;
}
}
import type { CeoHandler } from '../ceo.handler';
export async function updateUniqueSymbols(
this: CeoHandler,
_payload: unknown,
_context: any
): Promise<unknown> {
this.logger.info('Starting update to get unique CEO symbols by ceoId');
try {
// Get unique ceoId values from the ceoSymbols collection
const uniqueCeoIds = await this.mongodb.collection('ceoSymbols').distinct('ceoId');
this.logger.info(`Found ${uniqueCeoIds.length} unique CEO IDs`);
// Get detailed records for each unique ceoId (latest/first record)
const uniqueSymbols = [];
for (const ceoId of uniqueCeoIds) {
const symbol = await this.mongodb
.collection('ceoSymbols')
.findOne({ ceoId }, { sort: { _id: -1 } }); // Get latest record
if (symbol) {
uniqueSymbols.push(symbol);
}
}
this.logger.info(`Retrieved ${uniqueSymbols.length} unique symbol records`);
// Schedule individual jobs for each unique symbol
let scheduledJobs = 0;
for (const symbol of uniqueSymbols) {
// Schedule a job to process this individual symbol
await this.scheduleOperation('process-individual-symbol', {
ceoId: symbol.ceoId,
symbol: symbol.symbol,
});
scheduledJobs++;
// Add small delay to avoid overwhelming the queue
if (scheduledJobs % 10 === 0) {
this.logger.debug(`Scheduled ${scheduledJobs} jobs so far`);
}
}
this.logger.info(`Successfully scheduled ${scheduledJobs} individual symbol update jobs`);
// Cache the results for monitoring
await this.cacheSet(
'unique-symbols-last-run',
{
timestamp: new Date().toISOString(),
totalUniqueIds: uniqueCeoIds.length,
totalRecords: uniqueSymbols.length,
scheduledJobs,
},
1800
); // Cache for 30 minutes
return {
success: true,
uniqueCeoIds: uniqueCeoIds.length,
uniqueRecords: uniqueSymbols.length,
scheduledJobs,
timestamp: new Date().toISOString(),
};
} catch (error) {
this.logger.error('Failed to update unique CEO symbols', { error });
throw error;
}
}

26
apps/data-ingestion/src/handlers/ceo/ceo.handler.ts
View file

@ -3,13 +3,9 @@ import {
Handler,
Operation,
ScheduledOperation,
type IServiceContainer
type IServiceContainer,
} from '@stock-bot/handlers';
import {
processIndividualSymbol,
updateCeoChannels,
updateUniqueSymbols
} from './actions';
import { processIndividualSymbol, updateCeoChannels, updateUniqueSymbols } from './actions';
@Handler('ceo')
// @Disabled()
@ -18,21 +14,21 @@ export class CeoHandler extends BaseHandler {
super(services); // Handler name read from @Handler decorator
}
@ScheduledOperation('update-ceo-channels', '0 */15 * * *', {
priority: 7,
immediately: false,
description: 'Get all CEO symbols and exchanges'
@ScheduledOperation('update-ceo-channels', '0 */15 * * *', {
priority: 7,
immediately: false,
description: 'Get all CEO symbols and exchanges',
})
updateCeoChannels = updateCeoChannels;
@Operation('update-unique-symbols')
@ScheduledOperation('process-unique-symbols', '0 0 1 * *', {
priority: 5,
immediately: false,
description: 'Process unique CEO symbols and schedule individual jobs'
@ScheduledOperation('process-unique-symbols', '0 0 1 * *', {
priority: 5,
immediately: false,
description: 'Process unique CEO symbols and schedule individual jobs',
})
updateUniqueSymbols = updateUniqueSymbols;
@Operation('process-individual-symbol')
processIndividualSymbol = processIndividualSymbol;
}
}

221
apps/data-ingestion/src/handlers/example-handler.ts
View file

@ -1,114 +1,107 @@
import { OperationContext } from '@stock-bot/di';
import type { ServiceContainer } from '@stock-bot/di';
/**
* Example handler showing how to use the new connection pooling pattern
*/
export class ExampleHandler {
constructor(private readonly container: ServiceContainer) {}
/**
* Example operation using the enhanced OperationContext
*/
async performOperation(data: any): Promise<void> {
// Create operation context with container
const context = new OperationContext(
'example-handler',
'perform-operation',
this.container,
{ data }
);
try {
// Log operation start
context.logger.info('Starting operation', { data });
// Use MongoDB through service resolution
const mongodb = context.resolve<any>('mongodb');
const result = await mongodb.collection('test').insertOne(data);
context.logger.debug('MongoDB insert complete', { insertedId: result.insertedId });
// Use PostgreSQL through service resolution
const postgres = context.resolve<any>('postgres');
await postgres.query(
'INSERT INTO operations (id, status) VALUES ($1, $2)',
[result.insertedId, 'completed']
);
// Use cache through service resolution
const cache = context.resolve<any>('cache');
await cache.set(`operation:${result.insertedId}`, {
status: 'completed',
timestamp: new Date()
});
context.logger.info('Operation completed successfully');
} catch (error) {
context.logger.error('Operation failed', { error });
throw error;
}
}
/**
* Example of batch operation with isolated connection pool
*/
async performBatchOperation(items: any[]): Promise<void> {
// Create a scoped container for this batch operation
const scopedContainer = this.container.createScope();
const context = new OperationContext(
'example-handler',
'batch-operation',
scopedContainer,
{ itemCount: items.length }
);
try {
context.logger.info('Starting batch operation', { itemCount: items.length });
// Get services once for the batch
const mongodb = context.resolve<any>('mongodb');
const cache = context.resolve<any>('cache');
// Process items in parallel
const promises = items.map(async (item, index) => {
const itemContext = new OperationContext(
'example-handler',
`batch-item-${index}`,
scopedContainer,
{ item }
);
try {
await mongodb.collection('batch').insertOne(item);
await cache.set(`batch:${item.id}`, item);
} catch (error) {
itemContext.logger.error('Batch item failed', { error, itemIndex: index });
throw error;
}
});
await Promise.all(promises);
context.logger.info('Batch operation completed');
} finally {
// Clean up scoped resources
await scopedContainer.dispose();
}
}
}
/**
* Example of how to use in a job handler
*/
export async function createExampleJobHandler(container: ServiceContainer) {
return async (job: any) => {
const handler = new ExampleHandler(container);
if (job.data.type === 'batch') {
await handler.performBatchOperation(job.data.items);
} else {
await handler.performOperation(job.data);
}
};
}
import { OperationContext } from '@stock-bot/di';
import type { ServiceContainer } from '@stock-bot/di';
/**
* Example handler showing how to use the new connection pooling pattern
*/
export class ExampleHandler {
constructor(private readonly container: ServiceContainer) {}
/**
* Example operation using the enhanced OperationContext
*/
async performOperation(data: any): Promise<void> {
// Create operation context with container
const context = new OperationContext('example-handler', 'perform-operation', this.container, {
data,
});
try {
// Log operation start
context.logger.info('Starting operation', { data });
// Use MongoDB through service resolution
const mongodb = context.resolve<any>('mongodb');
const result = await mongodb.collection('test').insertOne(data);
context.logger.debug('MongoDB insert complete', { insertedId: result.insertedId });
// Use PostgreSQL through service resolution
const postgres = context.resolve<any>('postgres');
await postgres.query('INSERT INTO operations (id, status) VALUES ($1, $2)', [
result.insertedId,
'completed',
]);
// Use cache through service resolution
const cache = context.resolve<any>('cache');
await cache.set(`operation:${result.insertedId}`, {
status: 'completed',
timestamp: new Date(),
});
context.logger.info('Operation completed successfully');
} catch (error) {
context.logger.error('Operation failed', { error });
throw error;
}
}
/**
* Example of batch operation with isolated connection pool
*/
async performBatchOperation(items: any[]): Promise<void> {
// Create a scoped container for this batch operation
const scopedContainer = this.container.createScope();
const context = new OperationContext('example-handler', 'batch-operation', scopedContainer, {
itemCount: items.length,
});
try {
context.logger.info('Starting batch operation', { itemCount: items.length });
// Get services once for the batch
const mongodb = context.resolve<any>('mongodb');
const cache = context.resolve<any>('cache');
// Process items in parallel
const promises = items.map(async (item, index) => {
const itemContext = new OperationContext(
'example-handler',
`batch-item-${index}`,
scopedContainer,
{ item }
);
try {
await mongodb.collection('batch').insertOne(item);
await cache.set(`batch:${item.id}`, item);
} catch (error) {
itemContext.logger.error('Batch item failed', { error, itemIndex: index });
throw error;
}
});
await Promise.all(promises);
context.logger.info('Batch operation completed');
} finally {
// Clean up scoped resources
await scopedContainer.dispose();
}
}
}
/**
* Example of how to use in a job handler
*/
export async function createExampleJobHandler(container: ServiceContainer) {
return async (job: any) => {
const handler = new ExampleHandler(container);
if (job.data.type === 'batch') {
await handler.performBatchOperation(job.data.items);
} else {
await handler.performOperation(job.data);
}
};
}

188
apps/data-ingestion/src/handlers/example/example.handler.ts
View file

@ -1,94 +1,94 @@
/**
* Example Handler - Demonstrates ergonomic handler patterns
* Shows inline operations, service helpers, and scheduled operations
*/
import {
BaseHandler,
Handler,
Operation,
ScheduledOperation,
type ExecutionContext,
type IServiceContainer
} from '@stock-bot/handlers';
@Handler('example')
export class ExampleHandler extends BaseHandler {
constructor(services: IServiceContainer) {
super(services);
}
/**
* Simple inline operation - no separate action file needed
*/
@Operation('get-stats')
async getStats(): Promise<{ total: number; active: number; cached: boolean }> {
// Use collection helper for cleaner MongoDB access
const total = await this.collection('items').countDocuments();
const active = await this.collection('items').countDocuments({ status: 'active' });
// Use cache helpers with automatic prefixing
const cached = await this.cacheGet<number>('last-total');
await this.cacheSet('last-total', total, 300); // 5 minutes
// Use log helper with automatic handler context
this.log('info', 'Stats retrieved', { total, active });
return { total, active, cached: cached !== null };
}
/**
* Scheduled operation using combined decorator
*/
@ScheduledOperation('cleanup-old-items', '0 2 * * *', {
priority: 5,
description: 'Clean up items older than 30 days'
})
async cleanupOldItems(): Promise<{ deleted: number }> {
const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
const result = await this.collection('items').deleteMany({
createdAt: { $lt: thirtyDaysAgo }
});
this.log('info', 'Cleanup completed', { deleted: result.deletedCount });
// Schedule a follow-up task
await this.scheduleIn('generate-report', { type: 'cleanup' }, 60); // 1 minute
return { deleted: result.deletedCount };
}
/**
* Operation that uses proxy service
*/
@Operation('fetch-external-data')
async fetchExternalData(input: { url: string }): Promise<{ data: any }> {
const proxyUrl = this.proxy.getProxy();
if (!proxyUrl) {
throw new Error('No proxy available');
}
// Use HTTP client with proxy
const response = await this.http.get(input.url, {
proxy: proxyUrl,
timeout: 10000
});
// Cache the result
await this.cacheSet(`external:${input.url}`, response.data, 3600);
return { data: response.data };
}
/**
* Complex operation that still uses action file
*/
@Operation('process-batch')
async processBatch(input: any, context: ExecutionContext): Promise<unknown> {
// For complex operations, still use action files
const { processBatch } = await import('./actions/batch.action');
return processBatch(this, input);
}
}
/**
* Example Handler - Demonstrates ergonomic handler patterns
* Shows inline operations, service helpers, and scheduled operations
*/
import {
BaseHandler,
Handler,
Operation,
ScheduledOperation,
type ExecutionContext,
type IServiceContainer,
} from '@stock-bot/handlers';
@Handler('example')
export class ExampleHandler extends BaseHandler {
constructor(services: IServiceContainer) {
super(services);
}
/**
* Simple inline operation - no separate action file needed
*/
@Operation('get-stats')
async getStats(): Promise<{ total: number; active: number; cached: boolean }> {
// Use collection helper for cleaner MongoDB access
const total = await this.collection('items').countDocuments();
const active = await this.collection('items').countDocuments({ status: 'active' });
// Use cache helpers with automatic prefixing
const cached = await this.cacheGet<number>('last-total');
await this.cacheSet('last-total', total, 300); // 5 minutes
// Use log helper with automatic handler context
this.log('info', 'Stats retrieved', { total, active });
return { total, active, cached: cached !== null };
}
/**
* Scheduled operation using combined decorator
*/
@ScheduledOperation('cleanup-old-items', '0 2 * * *', {
priority: 5,
description: 'Clean up items older than 30 days',
})
async cleanupOldItems(): Promise<{ deleted: number }> {
const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
const result = await this.collection('items').deleteMany({
createdAt: { $lt: thirtyDaysAgo },
});
this.log('info', 'Cleanup completed', { deleted: result.deletedCount });
// Schedule a follow-up task
await this.scheduleIn('generate-report', { type: 'cleanup' }, 60); // 1 minute
return { deleted: result.deletedCount };
}
/**
* Operation that uses proxy service
*/
@Operation('fetch-external-data')
async fetchExternalData(input: { url: string }): Promise<{ data: any }> {
const proxyUrl = this.proxy.getProxy();
if (!proxyUrl) {
throw new Error('No proxy available');
}
// Use HTTP client with proxy
const response = await this.http.get(input.url, {
proxy: proxyUrl,
timeout: 10000,
});
// Cache the result
await this.cacheSet(`external:${input.url}`, response.data, 3600);
return { data: response.data };
}
/**
* Complex operation that still uses action file
*/
@Operation('process-batch')
async processBatch(input: any, context: ExecutionContext): Promise<unknown> {
// For complex operations, still use action files
const { processBatch } = await import('./actions/batch.action');
return processBatch(this, input);
}
}

38
apps/data-ingestion/src/handlers/ib/actions/fetch-exchanges-and-symbols.action.ts Normal file
View file

@ -0,0 +1,38 @@
import type { IbHandler } from '../ib.handler';
export async function fetchExchangesAndSymbols(this: IbHandler): Promise<unknown> {
this.logger.info('Starting IB exchanges and symbols fetch job');
try {
// Fetch session headers first
const sessionHeaders = await this.fetchSession();
if (!sessionHeaders) {
this.logger.error('Failed to get session headers for IB job');
return { success: false, error: 'No session headers' };
}
this.logger.info('Session headers obtained, fetching exchanges...');
// Fetch exchanges
const exchanges = await this.fetchExchanges();
this.logger.info('Fetched exchanges from IB', { count: exchanges?.length || 0 });
// Fetch symbols
this.logger.info('Fetching symbols...');
const symbols = await this.fetchSymbols();
this.logger.info('Fetched symbols from IB', { count: symbols?.length || 0 });
return {
success: true,
exchangesCount: exchanges?.length || 0,
symbolsCount: symbols?.length || 0,
};
} catch (error) {
this.logger.error('Failed to fetch IB exchanges and symbols', { error });
return {
success: false,
error: error instanceof Error ? error.message : 'Unknown error',
};
}
}

38
apps/data-ingestion/src/handlers/ib/operations/exchanges.operations.ts → apps/data-ingestion/src/handlers/ib/actions/fetch-exchanges.action.ts
View file

@ -1,16 +1,15 @@
/**
* IB Exchanges Operations - Fetching exchange data from IB API
*/
import { OperationContext } from '@stock-bot/di';
import type { ServiceContainer } from '@stock-bot/di';
import type { IbHandler } from '../ib.handler';
import { IB_CONFIG } from '../shared/config';
export async function fetchExchanges(sessionHeaders: Record<string, string>, container: ServiceContainer): Promise<unknown[] | null> {
const ctx = OperationContext.create('ib', 'exchanges', { container });
export async function fetchExchanges(this: IbHandler): Promise<unknown[] | null> {
try {
ctx.logger.info('🔍 Fetching exchanges with session headers...');
// First get session headers
const sessionHeaders = await this.fetchSession();
if (!sessionHeaders) {
throw new Error('Failed to get session headers');
}
this.logger.info('🔍 Fetching exchanges with session headers...');
// The URL for the exchange data API
const exchangeUrl = IB_CONFIG.BASE_URL + IB_CONFIG.EXCHANGE_API;
@ -28,7 +27,7 @@ export async function fetchExchanges(sessionHeaders: Record<string, string>, con
'X-Requested-With': 'XMLHttpRequest',
};
ctx.logger.info('📤 Making request to exchange API...', {
this.logger.info('📤 Making request to exchange API...', {
url: exchangeUrl,
headerCount: Object.keys(requestHeaders).length,
});
@ -41,7 +40,7 @@ export async function fetchExchanges(sessionHeaders: Record<string, string>, con
});
if (!response.ok) {
ctx.logger.error('❌ Exchange API request failed', {
this.logger.error('❌ Exchange API request failed', {
status: response.status,
statusText: response.statusText,
});
@ -50,19 +49,18 @@ export async function fetchExchanges(sessionHeaders: Record<string, string>, con
const data = await response.json();
const exchanges = data?.exchanges || [];
ctx.logger.info('✅ Exchange data fetched successfully');
this.logger.info('✅ Exchange data fetched successfully');
ctx.logger.info('Saving IB exchanges to MongoDB...');
await ctx.mongodb.batchUpsert('ibExchanges', exchanges, ['id', 'country_code']);
ctx.logger.info('✅ Exchange IB data saved to MongoDB:', {
this.logger.info('Saving IB exchanges to MongoDB...');
await this.mongodb.batchUpsert('ibExchanges', exchanges, ['id', 'country_code']);
this.logger.info('✅ Exchange IB data saved to MongoDB:', {
count: exchanges.length,
});
return exchanges;
} catch (error) {
ctx.logger.error('❌ Failed to fetch exchanges', { error });
this.logger.error('❌ Failed to fetch exchanges', { error });
return null;
} finally {
await ctx.dispose();
}
}
}

83
apps/data-ingestion/src/handlers/ib/actions/fetch-session.action.ts Normal file
View file

@ -0,0 +1,83 @@
import { Browser } from '@stock-bot/browser';
import type { IbHandler } from '../ib.handler';
import { IB_CONFIG } from '../shared/config';
export async function fetchSession(this: IbHandler): Promise<Record<string, string> | undefined> {
try {
await Browser.initialize({
headless: true,
timeout: IB_CONFIG.BROWSER_TIMEOUT,
blockResources: false,
});
this.logger.info('✅ Browser initialized');
const { page } = await Browser.createPageWithProxy(
IB_CONFIG.BASE_URL + IB_CONFIG.PRODUCTS_PAGE,
IB_CONFIG.DEFAULT_PROXY
);
this.logger.info('✅ Page created with proxy');
const headersPromise = new Promise<Record<string, string> | undefined>(resolve => {
let resolved = false;
page.onNetworkEvent(event => {
if (event.url.includes('/webrest/search/product-types/summary')) {
if (event.type === 'request') {
try {
resolve(event.headers);
} catch (e) {
resolve(undefined);
this.logger.debug('Raw Summary Response error', { error: (e as Error).message });
}
}
}
});
// Timeout fallback
setTimeout(() => {
if (!resolved) {
resolved = true;
this.logger.warn('Timeout waiting for headers');
resolve(undefined);
}
}, IB_CONFIG.HEADERS_TIMEOUT);
});
this.logger.info('⏳ Waiting for page load...');
await page.waitForLoadState('domcontentloaded', { timeout: IB_CONFIG.PAGE_LOAD_TIMEOUT });
this.logger.info('✅ Page loaded');
//Products tabs
this.logger.info('🔍 Looking for Products tab...');
const productsTab = page.locator('#productSearchTab[role="tab"][href="#products"]');
await productsTab.waitFor({ timeout: IB_CONFIG.ELEMENT_TIMEOUT });
this.logger.info('✅ Found Products tab');
this.logger.info('🖱️ Clicking Products tab...');
await productsTab.click();
this.logger.info('✅ Products tab clicked');
// New Products Checkbox
this.logger.info('🔍 Looking for "New Products Only" radio button...');
const radioButton = page.locator('span.checkbox-text:has-text("New Products Only")');
await radioButton.waitFor({ timeout: IB_CONFIG.ELEMENT_TIMEOUT });
this.logger.info(`🎯 Found "New Products Only" radio button`);
await radioButton.first().click();
this.logger.info('✅ "New Products Only" radio button clicked');
// Wait for and return headers immediately when captured
this.logger.info('⏳ Waiting for headers to be captured...');
const headers = await headersPromise;
page.close();
if (headers) {
this.logger.info('✅ Headers captured successfully');
} else {
this.logger.warn('⚠️ No headers were captured');
}
return headers;
} catch (error) {
this.logger.error('Failed to fetch IB symbol summary', { error });
return;
}
}

83
apps/data-ingestion/src/handlers/ib/operations/symbols.operations.ts → apps/data-ingestion/src/handlers/ib/actions/fetch-symbols.action.ts
View file

@ -1,18 +1,16 @@
/**
* IB Symbols Operations - Fetching symbol data from IB API
*/
import { OperationContext } from '@stock-bot/di';
import type { ServiceContainer } from '@stock-bot/di';
import type { IbHandler } from '../ib.handler';
import { IB_CONFIG } from '../shared/config';
// Fetch symbols from IB using the session headers
export async function fetchSymbols(sessionHeaders: Record<string, string>, container: ServiceContainer): Promise<unknown[] | null> {
const ctx = OperationContext.create('ib', 'symbols', { container });
export async function fetchSymbols(this: IbHandler): Promise<unknown[] | null> {
try {
ctx.logger.info('🔍 Fetching symbols with session headers...');
// First get session headers
const sessionHeaders = await this.fetchSession();
if (!sessionHeaders) {
throw new Error('Failed to get session headers');
}
this.logger.info('🔍 Fetching symbols with session headers...');
// Prepare headers - include all session headers plus any additional ones
const requestHeaders = {
...sessionHeaders,
@ -39,18 +37,15 @@ export async function fetchSymbols(sessionHeaders: Record<string, string>, conta
};
// Get Summary
const summaryResponse = await fetch(
IB_CONFIG.BASE_URL + IB_CONFIG.SUMMARY_API,
{
method: 'POST',
headers: requestHeaders,
proxy: IB_CONFIG.DEFAULT_PROXY,
body: JSON.stringify(requestBody),
}
);
const summaryResponse = await fetch(IB_CONFIG.BASE_URL + IB_CONFIG.SUMMARY_API, {
method: 'POST',
headers: requestHeaders,
proxy: IB_CONFIG.DEFAULT_PROXY,
body: JSON.stringify(requestBody),
});
if (!summaryResponse.ok) {
ctx.logger.error('❌ Summary API request failed', {
this.logger.error('❌ Summary API request failed', {
status: summaryResponse.status,
statusText: summaryResponse.statusText,
});
@ -58,36 +53,33 @@ export async function fetchSymbols(sessionHeaders: Record<string, string>, conta
}
const summaryData = await summaryResponse.json();
ctx.logger.info('✅ IB Summary data fetched successfully', {
this.logger.info('✅ IB Summary data fetched successfully', {
totalCount: summaryData[0].totalCount,
});
const symbols = [];
requestBody.pageSize = IB_CONFIG.PAGE_SIZE;
const pageCount = Math.ceil(summaryData[0].totalCount / IB_CONFIG.PAGE_SIZE) || 0;
ctx.logger.info('Fetching Symbols for IB', { pageCount });
this.logger.info('Fetching Symbols for IB', { pageCount });
const symbolPromises = [];
for (let page = 1; page <= pageCount; page++) {
requestBody.pageNumber = page;
// Fetch symbols for the current page
const symbolsResponse = fetch(
IB_CONFIG.BASE_URL + IB_CONFIG.PRODUCTS_API,
{
method: 'POST',
headers: requestHeaders,
proxy: IB_CONFIG.DEFAULT_PROXY,
body: JSON.stringify(requestBody),
}
);
const symbolsResponse = fetch(IB_CONFIG.BASE_URL + IB_CONFIG.PRODUCTS_API, {
method: 'POST',
headers: requestHeaders,
proxy: IB_CONFIG.DEFAULT_PROXY,
body: JSON.stringify(requestBody),
});
symbolPromises.push(symbolsResponse);
}
const responses = await Promise.all(symbolPromises);
for (const response of responses) {
if (!response.ok) {
ctx.logger.error('❌ Symbols API request failed', {
this.logger.error('❌ Symbols API request failed', {
status: response.status,
statusText: response.statusText,
});
@ -98,29 +90,28 @@ export async function fetchSymbols(sessionHeaders: Record<string, string>, conta
if (symJson && symJson.length > 0) {
symbols.push(...symJson);
} else {
ctx.logger.warn('⚠️ No symbols found in response');
this.logger.warn('⚠️ No symbols found in response');
continue;
}
}
if (symbols.length === 0) {
ctx.logger.warn('⚠️ No symbols fetched from IB');
this.logger.warn('⚠️ No symbols fetched from IB');
return null;
}
ctx.logger.info('✅ IB symbols fetched successfully, saving to DB...', {
this.logger.info('✅ IB symbols fetched successfully, saving to DB...', {
totalSymbols: symbols.length,
});
await ctx.mongodb.batchUpsert('ib_symbols', symbols, ['symbol', 'exchangeId']);
ctx.logger.info('Saved IB symbols to DB', {
await this.mongodb.batchUpsert('ib_symbols', symbols, ['symbol', 'exchangeId']);
this.logger.info('Saved IB symbols to DB', {
totalSymbols: symbols.length,
});
return symbols;
} catch (error) {
ctx.logger.error('❌ Failed to fetch symbols', { error });
this.logger.error('❌ Failed to fetch symbols', { error });
return null;
} finally {
await ctx.dispose();
}
}
}

5
apps/data-ingestion/src/handlers/ib/actions/index.ts Normal file
View file

@ -0,0 +1,5 @@
export { fetchSession } from './fetch-session.action';
export { fetchExchanges } from './fetch-exchanges.action';
export { fetchSymbols } from './fetch-symbols.action';
export { fetchExchangesAndSymbols } from './fetch-exchanges-and-symbols.action';

109
apps/data-ingestion/src/handlers/ib/ib.handler.ts
View file

@ -1,90 +1,33 @@
/**
* Interactive Brokers Provider for new queue system
*/
import { getLogger } from '@stock-bot/logger';
import {
createJobHandler,
handlerRegistry,
type HandlerConfigWithSchedule,
} from '@stock-bot/queue';
import type { ServiceContainer } from '@stock-bot/di';
BaseHandler,
Handler,
Operation,
ScheduledOperation,
type IServiceContainer,
} from '@stock-bot/handlers';
import { fetchExchanges, fetchExchangesAndSymbols, fetchSession, fetchSymbols } from './actions';
const logger = getLogger('ib-provider');
@Handler('ib')
export class IbHandler extends BaseHandler {
constructor(services: IServiceContainer) {
super(services);
}
// Initialize and register the IB provider
export function initializeIBProvider(container: ServiceContainer) {
logger.debug('Registering IB provider with scheduled jobs...');
@Operation('fetch-session')
fetchSession = fetchSession;
const ibProviderConfig: HandlerConfigWithSchedule = {
name: 'ib',
operations: {
'fetch-session': createJobHandler(async () => {
// payload contains session configuration (not used in current implementation)
logger.debug('Processing session fetch request');
const { fetchSession } = await import('./operations/session.operations');
return fetchSession(container);
}),
@Operation('fetch-exchanges')
fetchExchanges = fetchExchanges;
'fetch-exchanges': createJobHandler(async () => {
// payload should contain session headers
logger.debug('Processing exchanges fetch request');
const { fetchSession } = await import('./operations/session.operations');
const { fetchExchanges } = await import('./operations/exchanges.operations');
const sessionHeaders = await fetchSession(container);
if (sessionHeaders) {
return fetchExchanges(sessionHeaders, container);
}
throw new Error('Failed to get session headers');
}),
@Operation('fetch-symbols')
fetchSymbols = fetchSymbols;
'fetch-symbols': createJobHandler(async () => {
// payload should contain session headers
logger.debug('Processing symbols fetch request');
const { fetchSession } = await import('./operations/session.operations');
const { fetchSymbols } = await import('./operations/symbols.operations');
const sessionHeaders = await fetchSession(container);
if (sessionHeaders) {
return fetchSymbols(sessionHeaders, container);
}
throw new Error('Failed to get session headers');
}),
'ib-exchanges-and-symbols': createJobHandler(async () => {
// Legacy operation for scheduled jobs
logger.info('Fetching symbol summary from IB');
const { fetchSession } = await import('./operations/session.operations');
const { fetchExchanges } = await import('./operations/exchanges.operations');
const { fetchSymbols } = await import('./operations/symbols.operations');
const sessionHeaders = await fetchSession(container);
logger.info('Fetched symbol summary from IB');
if (sessionHeaders) {
logger.debug('Fetching exchanges from IB');
const exchanges = await fetchExchanges(sessionHeaders, container);
logger.info('Fetched exchanges from IB', { count: exchanges?.length });
logger.debug('Fetching symbols from IB');
const symbols = await fetchSymbols(sessionHeaders, container);
logger.info('Fetched symbols from IB', { symbols });
return { exchangesCount: exchanges?.length, symbolsCount: symbols?.length };
}
return null;
}),
},
scheduledJobs: [
{
type: 'ib-exchanges-and-symbols',
operation: 'ib-exchanges-and-symbols',
cronPattern: '0 0 * * 0', // Every Sunday at midnight
priority: 5,
description: 'Fetch and update IB exchanges and symbols data',
// immediately: true, // Don't run immediately during startup to avoid conflicts
},
],
};
handlerRegistry.registerWithSchedule(ibProviderConfig);
logger.debug('IB provider registered successfully with scheduled jobs');
@Operation('ib-exchanges-and-symbols')
@ScheduledOperation('ib-exchanges-and-symbols', '0 0 * * 0', {
priority: 5,
description: 'Fetch and update IB exchanges and symbols data',
immediately: false,
})
fetchExchangesAndSymbols = fetchExchangesAndSymbols;
}

91
apps/data-ingestion/src/handlers/ib/operations/session.operations.ts
View file

@ -1,91 +0,0 @@
/**
* IB Session Operations - Browser automation for session headers
*/
import { Browser } from '@stock-bot/browser';
import { OperationContext } from '@stock-bot/di';
import type { ServiceContainer } from '@stock-bot/di';
import { IB_CONFIG } from '../shared/config';
export async function fetchSession(container: ServiceContainer): Promise<Record<string, string> | undefined> {
const ctx = OperationContext.create('ib', 'session', { container });
try {
await Browser.initialize({
headless: true,
timeout: IB_CONFIG.BROWSER_TIMEOUT,
blockResources: false
});
ctx.logger.info('✅ Browser initialized');
const { page } = await Browser.createPageWithProxy(
IB_CONFIG.BASE_URL + IB_CONFIG.PRODUCTS_PAGE,
IB_CONFIG.DEFAULT_PROXY
);
ctx.logger.info('✅ Page created with proxy');
const headersPromise = new Promise<Record<string, string> | undefined>(resolve => {
let resolved = false;
page.onNetworkEvent(event => {
if (event.url.includes('/webrest/search/product-types/summary')) {
if (event.type === 'request') {
try {
resolve(event.headers);
} catch (e) {
resolve(undefined);
ctx.logger.debug('Raw Summary Response error', { error: (e as Error).message });
}
}
}
});
// Timeout fallback
setTimeout(() => {
if (!resolved) {
resolved = true;
ctx.logger.warn('Timeout waiting for headers');
resolve(undefined);
}
}, IB_CONFIG.HEADERS_TIMEOUT);
});
ctx.logger.info('⏳ Waiting for page load...');
await page.waitForLoadState('domcontentloaded', { timeout: IB_CONFIG.PAGE_LOAD_TIMEOUT });
ctx.logger.info('✅ Page loaded');
//Products tabs
ctx.logger.info('🔍 Looking for Products tab...');
const productsTab = page.locator('#productSearchTab[role=\"tab\"][href=\"#products\"]');
await productsTab.waitFor({ timeout: IB_CONFIG.ELEMENT_TIMEOUT });
ctx.logger.info('✅ Found Products tab');
ctx.logger.info('🖱️ Clicking Products tab...');
await productsTab.click();
ctx.logger.info('✅ Products tab clicked');
// New Products Checkbox
ctx.logger.info('🔍 Looking for \"New Products Only\" radio button...');
const radioButton = page.locator('span.checkbox-text:has-text(\"New Products Only\")');
await radioButton.waitFor({ timeout: IB_CONFIG.ELEMENT_TIMEOUT });
ctx.logger.info(`🎯 Found \"New Products Only\" radio button`);
await radioButton.first().click();
ctx.logger.info('✅ \"New Products Only\" radio button clicked');
// Wait for and return headers immediately when captured
ctx.logger.info('⏳ Waiting for headers to be captured...');
const headers = await headersPromise;
page.close();
if (headers) {
ctx.logger.info('✅ Headers captured successfully');
} else {
ctx.logger.warn('⚠️ No headers were captured');
}
return headers;
} catch (error) {
ctx.logger.error('Failed to fetch IB symbol summary', { error });
return;
} finally {
await ctx.dispose();
}
}

7
apps/data-ingestion/src/handlers/ib/shared/config.ts
View file

@ -8,16 +8,17 @@ export const IB_CONFIG = {
EXCHANGE_API: '/webrest/exchanges',
SUMMARY_API: '/webrest/search/product-types/summary',
PRODUCTS_API: '/webrest/search/products-by-filters',
// Browser configuration
BROWSER_TIMEOUT: 10000,
PAGE_LOAD_TIMEOUT: 20000,
ELEMENT_TIMEOUT: 5000,
HEADERS_TIMEOUT: 30000,
// API configuration
DEFAULT_PROXY: 'http://doimvbnb-US-rotate:w5fpiwrb9895@p.webshare.io:80',
PAGE_SIZE: 500,
PRODUCT_COUNTRIES: ['CA', 'US'],
PRODUCT_TYPES: ['STK'],
};
};

32
apps/data-ingestion/src/handlers/index.ts
View file

@ -6,11 +6,12 @@
import type { IServiceContainer } from '@stock-bot/handlers';
import { autoRegisterHandlers } from '@stock-bot/handlers';
import { getLogger } from '@stock-bot/logger';
// Import handlers for bundling (ensures they're included in the build)
import './qm/qm.handler';
import './webshare/webshare.handler';
import './ceo/ceo.handler';
import './ib/ib.handler';
// Add more handler imports as needed
const logger = getLogger('handler-init');
@ -21,21 +22,17 @@ const logger = getLogger('handler-init');
export async function initializeAllHandlers(serviceContainer: IServiceContainer): Promise<void> {
try {
// Auto-register all handlers in this directory
const result = await autoRegisterHandlers(
__dirname,
serviceContainer,
{
pattern: '.handler.',
exclude: ['test', 'spec'],
dryRun: false
}
);
const result = await autoRegisterHandlers(__dirname, serviceContainer, {
pattern: '.handler.',
exclude: ['test', 'spec'],
dryRun: false,
});
logger.info('Handler auto-registration complete', {
registered: result.registered,
failed: result.failed
failed: result.failed,
});
if (result.failed.length > 0) {
logger.error('Some handlers failed to register', { failed: result.failed });
}
@ -51,21 +48,20 @@ export async function initializeAllHandlers(serviceContainer: IServiceContainer)
*/
async function manualHandlerRegistration(serviceContainer: any): Promise<void> {
logger.warn('Falling back to manual handler registration');
try {
// // Import and register handlers manually
// const { QMHandler } = await import('./qm/qm.handler');
// const qmHandler = new QMHandler(serviceContainer);
// qmHandler.register();
// const { WebShareHandler } = await import('./webshare/webshare.handler');
// const webShareHandler = new WebShareHandler(serviceContainer);
// webShareHandler.register();
logger.info('Manual handler registration complete');
} catch (error) {
logger.error('Manual handler registration failed', { error });
throw error;
}
}
}

34
apps/data-ingestion/src/handlers/proxy/operations/check.operations.ts
View file

@ -1,24 +1,23 @@
/**
* Proxy Check Operations - Checking proxy functionality
*/
import type { ProxyInfo } from '@stock-bot/proxy';
import { OperationContext } from '@stock-bot/di';
import { getLogger } from '@stock-bot/logger';
import type { ProxyInfo } from '@stock-bot/proxy';
import { fetch } from '@stock-bot/utils';
import { PROXY_CONFIG } from '../shared/config';
/**
* Check if a proxy is working
*/
export async function checkProxy(proxy: ProxyInfo): Promise<ProxyInfo> {
const ctx = {
logger: getLogger('proxy-check'),
const ctx = {
logger: getLogger('proxy-check'),
resolve: <T>(_name: string) => {
throw new Error(`Service container not available for proxy operations`);
}
},
} as any;
let success = false;
ctx.logger.debug(`Checking Proxy:`, {
protocol: proxy.protocol,
@ -28,16 +27,17 @@ export async function checkProxy(proxy: ProxyInfo): Promise<ProxyInfo> {
try {
// Test the proxy using fetch with proxy support
const proxyUrl = proxy.username && proxy.password
? `${proxy.protocol}://${encodeURIComponent(proxy.username)}:${encodeURIComponent(proxy.password)}@${proxy.host}:${proxy.port}`
: `${proxy.protocol}://${proxy.host}:${proxy.port}`;
const proxyUrl =
proxy.username && proxy.password
? `${proxy.protocol}://${encodeURIComponent(proxy.username)}:${encodeURIComponent(proxy.password)}@${proxy.host}:${proxy.port}`
: `${proxy.protocol}://${proxy.host}:${proxy.port}`;
const response = await fetch(PROXY_CONFIG.CHECK_URL, {
proxy: proxyUrl,
signal: AbortSignal.timeout(PROXY_CONFIG.CHECK_TIMEOUT),
logger: ctx.logger
logger: ctx.logger,
} as any);
const data = await response.text();
const isWorking = response.ok;
@ -94,7 +94,11 @@ export async function checkProxy(proxy: ProxyInfo): Promise<ProxyInfo> {
/**
* Update proxy data in cache with working/total stats and average response time
*/
async function updateProxyInCache(proxy: ProxyInfo, isWorking: boolean, ctx: OperationContext): Promise<void> {
async function updateProxyInCache(
proxy: ProxyInfo,
isWorking: boolean,
ctx: OperationContext
): Promise<void> {
const _cacheKey = `${PROXY_CONFIG.CACHE_KEY}:${proxy.protocol}://${proxy.host}:${proxy.port}`;
try {
@ -167,6 +171,6 @@ async function updateProxyInCache(proxy: ProxyInfo, isWorking: boolean, ctx: Ope
function updateProxyStats(sourceId: string, success: boolean, ctx: OperationContext) {
// Stats are now handled by the global ProxyManager
ctx.logger.debug('Proxy check result', { sourceId, success });
// TODO: Integrate with global ProxyManager stats if needed
}
}

12
apps/data-ingestion/src/handlers/proxy/operations/query.operations.ts
View file

@ -1,9 +1,8 @@
/**
* Proxy Query Operations - Getting active proxies from cache
*/
import type { ProxyInfo } from '@stock-bot/proxy';
import { OperationContext } from '@stock-bot/di';
import type { ProxyInfo } from '@stock-bot/proxy';
import { PROXY_CONFIG } from '../shared/config';
/**
@ -17,7 +16,7 @@ export async function getRandomActiveProxy(
minSuccessRate: number = 50
): Promise<ProxyInfo | null> {
const ctx = OperationContext.create('proxy', 'get-random');
try {
// Get all active proxy keys from cache
const pattern = protocol
@ -56,7 +55,10 @@ export async function getRandomActiveProxy(
return proxyData;
}
} catch (error) {
ctx.logger.debug('Error reading proxy from cache', { key, error: (error as Error).message });
ctx.logger.debug('Error reading proxy from cache', {
key,
error: (error as Error).message,
});
continue;
}
}
@ -76,4 +78,4 @@ export async function getRandomActiveProxy(
});
return null;
}
}
}

8
apps/data-ingestion/src/handlers/proxy/operations/queue.operations.ts
View file

@ -1,13 +1,13 @@
/**
* Proxy Queue Operations - Queueing proxy operations
*/
import { OperationContext } from '@stock-bot/di';
import type { ProxyInfo } from '@stock-bot/proxy';
import { QueueManager } from '@stock-bot/queue';
import { OperationContext } from '@stock-bot/di';
export async function queueProxyFetch(): Promise<string> {
const ctx = OperationContext.create('proxy', 'queue-fetch');
const queueManager = QueueManager.getInstance();
const queue = queueManager.getQueue('proxy');
const job = await queue.add('proxy-fetch', {
@ -24,7 +24,7 @@ export async function queueProxyFetch(): Promise<string> {
export async function queueProxyCheck(proxies: ProxyInfo[]): Promise<string> {
const ctx = OperationContext.create('proxy', 'queue-check');
const queueManager = QueueManager.getInstance();
const queue = queueManager.getQueue('proxy');
const job = await queue.add('proxy-check', {
@ -37,4 +37,4 @@ export async function queueProxyCheck(proxies: ProxyInfo[]): Promise<string> {
const jobId = job.id || 'unknown';
ctx.logger.info('Proxy check job queued', { jobId, count: proxies.length });
return jobId;
}
}

10
apps/data-ingestion/src/handlers/proxy/proxy.handler.ts
View file

@ -1,10 +1,14 @@
/**
* Proxy Provider for new queue system
*/
import type { ProxyInfo } from '@stock-bot/proxy';
import { getLogger } from '@stock-bot/logger';
import { handlerRegistry, createJobHandler, type HandlerConfigWithSchedule } from '@stock-bot/queue';
import type { ServiceContainer } from '@stock-bot/di';
import { getLogger } from '@stock-bot/logger';
import type { ProxyInfo } from '@stock-bot/proxy';
import {
createJobHandler,
handlerRegistry,
type HandlerConfigWithSchedule,
} from '@stock-bot/queue';
const handlerLogger = getLogger('proxy-handler');

2
apps/data-ingestion/src/handlers/proxy/shared/config.ts
View file

@ -137,4 +137,4 @@ export const PROXY_CONFIG = {
protocol: 'https',
},
],
};
};

2
apps/data-ingestion/src/handlers/proxy/shared/types.ts
View file

@ -10,4 +10,4 @@ export interface ProxySource {
total?: number; // Optional, used for stats
percentWorking?: number; // Optional, used for stats
lastChecked?: Date; // Optional, used for stats
}
}

12
apps/data-ingestion/src/handlers/qm/actions/exchanges.action.ts
View file

@ -6,16 +6,14 @@ import type { IServiceContainer } from '@stock-bot/handlers';
export async function fetchExchanges(services: IServiceContainer): Promise<any[]> {
// Get exchanges from MongoDB
const exchanges = await services.mongodb.collection('qm_exchanges')
.find({}).toArray();
const exchanges = await services.mongodb.collection('qm_exchanges').find({}).toArray();
return exchanges;
}
export async function getExchangeByCode(services: IServiceContainer, code: string): Promise<any> {
// Get specific exchange by code
const exchange = await services.mongodb.collection('qm_exchanges')
.findOne({ code });
const exchange = await services.mongodb.collection('qm_exchanges').findOne({ code });
return exchange;
}
}

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

@ -9,10 +9,10 @@ import { QMSessionManager } from '../shared/session-manager';
/**
* Check existing sessions and queue creation jobs for needed sessions
*/
export async function checkSessions(handler: BaseHandler): Promise<{
cleaned: number;
queued: number;
message: string;
export async function checkSessions(handler: BaseHandler): Promise<{
cleaned: number;
queued: number;
message: string;
}> {
const sessionManager = QMSessionManager.getInstance();
const cleanedCount = sessionManager.cleanupFailedSessions();
@ -24,17 +24,17 @@ export async function checkSessions(handler: BaseHandler): Promise<{
const currentCount = sessionManager.getSessions(sessionId).length;
const neededSessions = SESSION_CONFIG.MAX_SESSIONS - currentCount;
for (let i = 0; i < neededSessions; i++) {
await handler.scheduleOperation('create-session', { sessionId , sessionType });
await handler.scheduleOperation('create-session', { sessionId, sessionType });
handler.logger.info(`Queued job to create session for ${sessionType}`);
queuedCount++;
}
}
}
return {
cleaned: cleanedCount,
queued: queuedCount,
message: `Session check completed: cleaned ${cleanedCount}, queued ${queuedCount}`
message: `Session check completed: cleaned ${cleanedCount}, queued ${queuedCount}`,
};
}
@ -42,16 +42,15 @@ export async function checkSessions(handler: BaseHandler): Promise<{
* Create a single session for a specific session ID
*/
export async function createSingleSession(
handler: BaseHandler,
handler: BaseHandler,
input: any
): Promise<{ sessionId: string; status: string; sessionType: string }> {
const { sessionId, sessionType } = input || {};
const sessionManager = QMSessionManager.getInstance();
// Get proxy from proxy service
const proxyString = handler.proxy.getProxy();
// const session = {
// proxy: proxyString || 'http://proxy:8080',
// headers: sessionManager.getQmHeaders(),
@ -60,15 +59,14 @@ export async function createSingleSession(
// lastUsed: new Date()
// };
handler.logger.info(`Creating session for ${sessionType}`)
handler.logger.info(`Creating session for ${sessionType}`);
// Add session to manager
// sessionManager.addSession(sessionType, session);
return {
sessionId: sessionType,
status: 'created',
sessionType
sessionType,
};
}

9
apps/data-ingestion/src/handlers/qm/actions/spider.action.ts
View file

@ -9,16 +9,15 @@ export async function spiderSymbolSearch(
services: IServiceContainer,
config: SymbolSpiderJob
): Promise<{ foundSymbols: number; depth: number }> {
// Simple spider implementation
// TODO: Implement actual API calls to discover symbols
// For now, just return mock results
const foundSymbols = Math.floor(Math.random() * 10) + 1;
return {
foundSymbols,
depth: config.depth
depth: config.depth,
};
}
@ -31,4 +30,4 @@ export async function queueSymbolDiscovery(
// TODO: Queue actual discovery jobs
await services.cache.set(`discovery:${term}`, { queued: true }, 3600);
}
}
}

12
apps/data-ingestion/src/handlers/qm/actions/symbols.action.ts
View file

@ -6,16 +6,14 @@ import type { IServiceContainer } from '@stock-bot/handlers';
export async function searchSymbols(services: IServiceContainer): Promise<any[]> {
// Get symbols from MongoDB
const symbols = await services.mongodb.collection('qm_symbols')
.find({}).limit(50).toArray();
const symbols = await services.mongodb.collection('qm_symbols').find({}).limit(50).toArray();
return symbols;
}
export async function fetchSymbolData(services: IServiceContainer, symbol: string): Promise<any> {
// Fetch data for a specific symbol
const symbolData = await services.mongodb.collection('qm_symbols')
.findOne({ symbol });
const symbolData = await services.mongodb.collection('qm_symbols').findOne({ symbol });
return symbolData;
}
}

45
apps/data-ingestion/src/handlers/qm/qm.handler.ts
View file

@ -1,8 +1,4 @@
import {
BaseHandler,
Handler,
type IServiceContainer
} from '@stock-bot/handlers';
import { BaseHandler, Handler, type IServiceContainer } from '@stock-bot/handlers';
@Handler('qm')
export class QMHandler extends BaseHandler {
@ -11,10 +7,10 @@ export class QMHandler extends BaseHandler {
}
// @Operation('check-sessions')
// @QueueSchedule('0 */15 * * *', {
// priority: 7,
// immediately: true,
// description: 'Check and maintain QM sessions'
// @QueueSchedule('0 */15 * * *', {
// priority: 7,
// immediately: true,
// description: 'Check and maintain QM sessions'
// })
// async checkSessions(input: unknown, context: ExecutionContext): Promise<unknown> {
// // Call the session maintenance action
@ -36,13 +32,13 @@ export class QMHandler extends BaseHandler {
// // Check existing symbols in MongoDB
// const symbolsCollection = this.mongodb.collection('qm_symbols');
// const symbols = await symbolsCollection.find({}).limit(100).toArray();
// this.logger.info('QM symbol search completed', { count: symbols.length });
// if (symbols && symbols.length > 0) {
// // Cache result for performance
// await this.cache.set('qm-symbols-sample', symbols.slice(0, 10), 1800);
// return {
// success: true,
// message: 'QM symbol search completed successfully',
@ -58,7 +54,7 @@ export class QMHandler extends BaseHandler {
// count: 0,
// };
// }
// } catch (error) {
// this.logger.error('Failed to search QM symbols', { error });
// throw error;
@ -66,10 +62,10 @@ export class QMHandler extends BaseHandler {
// }
// @Operation('spider-symbol-search')
// @QueueSchedule('0 0 * * 0', {
// priority: 10,
// immediately: false,
// description: 'Comprehensive symbol search using QM API'
// @QueueSchedule('0 0 * * 0', {
// priority: 10,
// immediately: false,
// description: 'Comprehensive symbol search using QM API'
// })
// async spiderSymbolSearch(payload: SymbolSpiderJob | undefined, context: ExecutionContext): Promise<unknown> {
// // Set default payload for scheduled runs
@ -79,9 +75,9 @@ export class QMHandler extends BaseHandler {
// source: 'qm',
// maxDepth: 4
// };
// this.logger.info('Starting QM spider symbol search', { payload: jobPayload });
// // Store spider job info in cache (temporary data)
// const spiderJobId = `spider:qm:${Date.now()}:${Math.random().toString(36).substr(2, 9)}`;
// const spiderResult = {
@ -90,19 +86,18 @@ export class QMHandler extends BaseHandler {
// status: 'started',
// jobId: spiderJobId
// };
// // Store in cache with 1 hour TTL (temporary data)
// await this.cache.set(spiderJobId, spiderResult, 3600);
// this.logger.debug('Spider job stored in cache', { spiderJobId, ttl: 3600 });
// // Schedule follow-up processing if needed
// await this.scheduleOperation('search-symbols', { source: 'spider', spiderJobId }, 5000);
// return {
// success: true,
// return {
// success: true,
// message: 'QM spider search initiated',
// spiderJobId
// };
// }
}

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

@ -2,11 +2,10 @@
* Shared configuration for QM operations
*/
// QM Session IDs for different endpoints
export const QM_SESSION_IDS = {
LOOKUP: 'dc8c9930437f65d30f6597768800957017bac203a0a50342932757c8dfa158d6', // lookup endpoint
// '5ad521e05faf5778d567f6d0012ec34d6cdbaeb2462f41568f66558bc7b4ced9': [], //4488d072b
// '5ad521e05faf5778d567f6d0012ec34d6cdbaeb2462f41568f66558bc7b4ced9': [], //4488d072b
// cc1cbdaf040f76db8f4c94f7d156b9b9b716e1a7509ec9c74a48a47f6b6b9f87: [], //97ff00cf3 // getQuotes
// '74963ff42f1db2320d051762b5d3950ff9eab23f9d5c5b592551b4ca0441d086': [], //32ca24e394b // getSplitsBySymbol getBrokerRatingsBySymbol getDividendsBySymbol getEarningsSurprisesBySymbol getEarningsEventsBySymbol
// '1e1d7cb1de1fd2fe52684abdea41a446919a5fe12776dfab88615ac1ce1ec2f6': [], //fb5721812d2c // getEnhancedQuotes getProfiles
@ -36,4 +35,4 @@ export const SESSION_CONFIG = {
MAX_FAILED_CALLS: 10,
SESSION_TIMEOUT: 10000, // 10 seconds
API_TIMEOUT: 15000, // 15 seconds
} as const;
} as const;

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

@ -33,13 +33,15 @@ export class QMSessionManager {
if (!sessions || sessions.length === 0) {
return null;
}
// Filter out sessions with excessive failures
const validSessions = sessions.filter(session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS);
const validSessions = sessions.filter(
session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS
);
if (validSessions.length === 0) {
return null;
}
return validSessions[Math.floor(Math.random() * validSessions.length)];
}
@ -72,7 +74,7 @@ export class QMSessionManager {
*/
cleanupFailedSessions(): number {
let removedCount = 0;
Object.keys(this.sessionCache).forEach(sessionId => {
const initialCount = this.sessionCache[sessionId].length;
this.sessionCache[sessionId] = this.sessionCache[sessionId].filter(
@ -80,7 +82,7 @@ export class QMSessionManager {
);
removedCount += initialCount - this.sessionCache[sessionId].length;
});
return removedCount;
}
@ -94,13 +96,15 @@ export class QMSessionManager {
Referer: 'https://www.quotemedia.com/',
};
}
/**
* Check if more sessions are needed for a session ID
*/
needsMoreSessions(sessionId: string): boolean {
const sessions = this.sessionCache[sessionId] || [];
const validSessions = sessions.filter(session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS);
const validSessions = sessions.filter(
session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS
);
return validSessions.length < SESSION_CONFIG.MIN_SESSIONS;
}
@ -117,18 +121,22 @@ export class QMSessionManager {
*/
getStats() {
const stats: Record<string, { total: number; valid: number; failed: number }> = {};
Object.entries(this.sessionCache).forEach(([sessionId, sessions]) => {
const validSessions = sessions.filter(session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS);
const failedSessions = sessions.filter(session => session.failedCalls > SESSION_CONFIG.MAX_FAILED_CALLS);
const validSessions = sessions.filter(
session => session.failedCalls <= SESSION_CONFIG.MAX_FAILED_CALLS
);
const failedSessions = sessions.filter(
session => session.failedCalls > SESSION_CONFIG.MAX_FAILED_CALLS
);
stats[sessionId] = {
total: sessions.length,
valid: validSessions.length,
failed: failedSessions.length
failed: failedSessions.length,
};
});
return stats;
}
@ -145,4 +153,4 @@ export class QMSessionManager {
getInitialized(): boolean {
return this.isInitialized;
}
}
}

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

@ -29,4 +29,4 @@ export interface SpiderResult {
success: boolean;
symbolsFound: number;
jobsCreated: number;
}
}

55
apps/data-ingestion/src/handlers/webshare/operations/fetch.operations.ts
View file

@ -1,9 +1,8 @@
/**
* WebShare Fetch Operations - API integration
*/
import type { ProxyInfo } from '@stock-bot/proxy';
import { OperationContext } from '@stock-bot/di';
import type { ProxyInfo } from '@stock-bot/proxy';
import { WEBSHARE_CONFIG } from '../shared/config';
/**
@ -11,7 +10,7 @@ import { WEBSHARE_CONFIG } from '../shared/config';
*/
export async function fetchWebShareProxies(): Promise<ProxyInfo[]> {
const ctx = OperationContext.create('webshare', 'fetch-proxies');
try {
// Get configuration from config system
const { getConfig } = await import('@stock-bot/config');
@ -30,14 +29,17 @@ export async function fetchWebShareProxies(): Promise<ProxyInfo[]> {
ctx.logger.info('Fetching proxies from WebShare API', { apiUrl });
const response = await fetch(`${apiUrl}proxy/list/?mode=${WEBSHARE_CONFIG.DEFAULT_MODE}&page=${WEBSHARE_CONFIG.DEFAULT_PAGE}&page_size=${WEBSHARE_CONFIG.DEFAULT_PAGE_SIZE}`, {
method: 'GET',
headers: {
Authorization: `Token ${apiKey}`,
'Content-Type': 'application/json',
},
signal: AbortSignal.timeout(WEBSHARE_CONFIG.TIMEOUT),
});
const response = await fetch(
`${apiUrl}proxy/list/?mode=${WEBSHARE_CONFIG.DEFAULT_MODE}&page=${WEBSHARE_CONFIG.DEFAULT_PAGE}&page_size=${WEBSHARE_CONFIG.DEFAULT_PAGE_SIZE}`,
{
method: 'GET',
headers: {
Authorization: `Token ${apiKey}`,
'Content-Type': 'application/json',
},
signal: AbortSignal.timeout(WEBSHARE_CONFIG.TIMEOUT),
}
);
if (!response.ok) {
ctx.logger.error('WebShare API request failed', {
@ -55,22 +57,19 @@ export async function fetchWebShareProxies(): Promise<ProxyInfo[]> {
}
// 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, // WebShare provides working proxies
firstSeen: new Date(),
lastChecked: new Date(),
}));
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, // WebShare provides working proxies
firstSeen: new Date(),
lastChecked: new Date(),
})
);
ctx.logger.info('Successfully fetched proxies from WebShare', {
count: proxies.length,
@ -82,4 +81,4 @@ export async function fetchWebShareProxies(): Promise<ProxyInfo[]> {
ctx.logger.error('Failed to fetch proxies from WebShare', { error });
return [];
}
}
}

2
apps/data-ingestion/src/handlers/webshare/shared/config.ts
View file

@ -7,4 +7,4 @@ export const WEBSHARE_CONFIG = {
DEFAULT_MODE: 'direct',
DEFAULT_PAGE: 1,
TIMEOUT: 10000,
};
};

39
apps/data-ingestion/src/handlers/webshare/webshare.handler.ts
View file

@ -4,7 +4,7 @@ import {
Operation,
QueueSchedule,
type ExecutionContext,
type IServiceContainer
type IServiceContainer,
} from '@stock-bot/handlers';
@Handler('webshare')
@ -14,33 +14,45 @@ export class WebShareHandler extends BaseHandler {
}
@Operation('fetch-proxies')
@QueueSchedule('0 */6 * * *', {
priority: 3,
immediately: true,
description: 'Fetch fresh proxies from WebShare API'
@QueueSchedule('0 */6 * * *', {
priority: 3,
immediately: true,
description: 'Fetch fresh proxies from WebShare API',
})
async fetchProxies(_input: unknown, _context: ExecutionContext): Promise<unknown> {
this.logger.info('Fetching proxies from WebShare API');
try {
const { fetchWebShareProxies } = await import('./operations/fetch.operations');
const proxies = await fetchWebShareProxies();
if (proxies.length > 0) {
// Update the centralized proxy manager using the injected service
if (!this.proxy) {
this.logger.warn('Proxy manager is not initialized, cannot update proxies');
return {
success: false,
proxiesUpdated: 0,
error: 'Proxy manager not initialized',
};
}
await this.proxy.updateProxies(proxies);
this.logger.info('Updated proxy manager with WebShare proxies', {
this.logger.info('Updated proxy manager with WebShare proxies', {
count: proxies.length,
workingCount: proxies.filter(p => p.isWorking !== false).length,
});
// Cache proxy stats for monitoring
await this.cache.set('webshare-proxy-count', proxies.length, 3600);
await this.cache.set('webshare-working-count', proxies.filter(p => p.isWorking !== false).length, 3600);
await this.cache.set(
'webshare-working-count',
proxies.filter(p => p.isWorking !== false).length,
3600
);
await this.cache.set('last-webshare-fetch', new Date().toISOString(), 1800);
return {
return {
success: true,
proxiesUpdated: proxies.length,
workingProxies: proxies.filter(p => p.isWorking !== false).length,
@ -59,4 +71,3 @@ export class WebShareHandler extends BaseHandler {
}
}
}

44
apps/data-ingestion/src/index.ts
View file

@ -4,20 +4,18 @@
*/
// Framework imports
import { initializeServiceConfig } from '@stock-bot/config';
import { Hono } from 'hono';
import { cors } from 'hono/cors';
import { initializeServiceConfig } from '@stock-bot/config';
// Library imports
import {
createServiceContainer,
initializeServices as initializeAwilixServices,
type ServiceContainer
type ServiceContainer,
} from '@stock-bot/di';
import { getLogger, setLoggerConfig, shutdownLoggers } from '@stock-bot/logger';
import { Shutdown } from '@stock-bot/shutdown';
import { handlerRegistry } from '@stock-bot/types';
// Local imports
import { createRoutes } from './routes/create-routes';
import { initializeAllHandlers } from './handlers';
@ -84,17 +82,17 @@ async function initializeServices() {
ttl: 3600,
},
};
container = createServiceContainer(awilixConfig);
await initializeAwilixServices(container);
logger.info('Awilix container created and initialized');
// Get the service container for handlers
const serviceContainer = container.resolve('serviceContainer');
// Create app with routes
app = new Hono();
// Add CORS middleware
app.use(
'*',
@ -105,17 +103,17 @@ async function initializeServices() {
credentials: false,
})
);
// Create and mount routes using the service container
const routes = createRoutes(serviceContainer);
app.route('/', routes);
// Initialize handlers with service container from Awilix
logger.debug('Initializing data handlers with Awilix DI pattern...');
// Auto-register all handlers with the service container from Awilix
await initializeAllHandlers(serviceContainer);
logger.info('Data handlers initialized with new DI pattern');
// Create scheduled jobs from registered handlers
@ -175,10 +173,10 @@ async function initializeServices() {
logger.info('All services initialized successfully');
} catch (error) {
console.error('DETAILED ERROR:', error);
logger.error('Failed to initialize services', {
logger.error('Failed to initialize services', {
error: error instanceof Error ? error.message : String(error),
stack: error instanceof Error ? error.stack : undefined,
details: JSON.stringify(error, null, 2)
details: JSON.stringify(error, null, 2),
});
throw error;
}
@ -236,14 +234,20 @@ shutdown.onShutdownMedium(async () => {
if (container) {
// Disconnect database clients
const mongoClient = container.resolve('mongoClient');
if (mongoClient?.disconnect) await mongoClient.disconnect();
if (mongoClient?.disconnect) {
await mongoClient.disconnect();
}
const postgresClient = container.resolve('postgresClient');
if (postgresClient?.disconnect) await postgresClient.disconnect();
if (postgresClient?.disconnect) {
await postgresClient.disconnect();
}
const questdbClient = container.resolve('questdbClient');
if (questdbClient?.disconnect) await questdbClient.disconnect();
if (questdbClient?.disconnect) {
await questdbClient.disconnect();
}
logger.info('All services disposed successfully');
}
} catch (error) {
@ -268,4 +272,4 @@ startServer().catch(error => {
process.exit(1);
});
logger.info('Data service startup initiated with improved DI pattern');
logger.info('Data service startup initiated with improved DI pattern');

143
apps/data-ingestion/src/routes/create-routes.ts
View file

@ -1,69 +1,74 @@
/**
* Routes creation with improved DI pattern
*/
import { Hono } from 'hono';
import type { IServiceContainer } from '@stock-bot/handlers';
import { exchangeRoutes } from './exchange.routes';
import { healthRoutes } from './health.routes';
import { queueRoutes } from './queue.routes';
/**
* Creates all routes with access to type-safe services
*/
export function createRoutes(services: IServiceContainer): Hono {
const app = new Hono();
// Mount routes that don't need services
app.route('/health', healthRoutes);
// Mount routes that need services (will be updated to use services)
app.route('/api/exchanges', exchangeRoutes);
app.route('/api/queue', queueRoutes);
// Store services in app context for handlers that need it
app.use('*', async (c, next) => {
c.set('services', services);
await next();
});
// Add a new endpoint to test the improved DI
app.get('/api/di-test', async (c) => {
try {
const services = c.get('services') as IServiceContainer;
// Test MongoDB connection
const mongoStats = services.mongodb?.getPoolMetrics?.() || { status: services.mongodb ? 'connected' : 'disabled' };
// Test PostgreSQL connection
const pgConnected = services.postgres?.connected || false;
// Test cache
const cacheReady = services.cache?.isReady() || false;
// Test queue
const queueStats = services.queue?.getGlobalStats() || { status: 'disabled' };
return c.json({
success: true,
message: 'Improved DI pattern is working!',
services: {
mongodb: mongoStats,
postgres: { connected: pgConnected },
cache: { ready: cacheReady },
queue: queueStats
},
timestamp: new Date().toISOString()
});
} catch (error) {
const services = c.get('services') as IServiceContainer;
services.logger.error('DI test endpoint failed', { error });
return c.json({
success: false,
error: error instanceof Error ? error.message : String(error)
}, 500);
}
});
return app;
}
/**
* Routes creation with improved DI pattern
*/
import { Hono } from 'hono';
import type { IServiceContainer } from '@stock-bot/handlers';
import { exchangeRoutes } from './exchange.routes';
import { healthRoutes } from './health.routes';
import { queueRoutes } from './queue.routes';
/**
* Creates all routes with access to type-safe services
*/
export function createRoutes(services: IServiceContainer): Hono {
const app = new Hono();
// Mount routes that don't need services
app.route('/health', healthRoutes);
// Mount routes that need services (will be updated to use services)
app.route('/api/exchanges', exchangeRoutes);
app.route('/api/queue', queueRoutes);
// Store services in app context for handlers that need it
app.use('*', async (c, next) => {
c.set('services', services);
await next();
});
// Add a new endpoint to test the improved DI
app.get('/api/di-test', async c => {
try {
const services = c.get('services') as IServiceContainer;
// Test MongoDB connection
const mongoStats = services.mongodb?.getPoolMetrics?.() || {
status: services.mongodb ? 'connected' : 'disabled',
};
// Test PostgreSQL connection
const pgConnected = services.postgres?.connected || false;
// Test cache
const cacheReady = services.cache?.isReady() || false;
// Test queue
const queueStats = services.queue?.getGlobalStats() || { status: 'disabled' };
return c.json({
success: true,
message: 'Improved DI pattern is working!',
services: {
mongodb: mongoStats,
postgres: { connected: pgConnected },
cache: { ready: cacheReady },
queue: queueStats,
},
timestamp: new Date().toISOString(),
});
} catch (error) {
const services = c.get('services') as IServiceContainer;
services.logger.error('DI test endpoint failed', { error });
return c.json(
{
success: false,
error: error instanceof Error ? error.message : String(error),
},
500
);
}
});
return app;
}

4
apps/data-ingestion/src/routes/exchange.routes.ts
View file

@ -11,7 +11,7 @@ exchange.get('/', async c => {
return c.json({
status: 'success',
data: [],
message: 'Exchange endpoints will be implemented with database integration'
message: 'Exchange endpoints will be implemented with database integration',
});
} catch (error) {
logger.error('Failed to get exchanges', { error });
@ -19,4 +19,4 @@ exchange.get('/', async c => {
}
});
export { exchange as exchangeRoutes };
export { exchange as exchangeRoutes };

6
apps/data-ingestion/src/routes/queue.routes.ts
View file

@ -10,11 +10,11 @@ queue.get('/status', async c => {
try {
const queueManager = QueueManager.getInstance();
const globalStats = await queueManager.getGlobalStats();
return c.json({
status: 'success',
data: globalStats,
message: 'Queue status retrieved successfully'
message: 'Queue status retrieved successfully',
});
} catch (error) {
logger.error('Failed to get queue status', { error });
@ -22,4 +22,4 @@ queue.get('/status', async c => {
}
});
export { queue as queueRoutes };
export { queue as queueRoutes };

2
apps/data-ingestion/src/types/exchange.types.ts
View file

@ -37,4 +37,4 @@ export interface IBSymbol {
name?: string;
currency?: string;
// Add other properties as needed
}
}

2
apps/data-ingestion/src/types/job-payloads.ts
View file

@ -90,4 +90,4 @@ export interface FetchWebShareProxiesResult extends CountableJobResult {
// No payload job types (for operations that don't need input)
export interface NoPayload {
// Empty interface for operations that don't need payload
}
}

2
apps/data-ingestion/src/utils/symbol-search.util.ts
View file

@ -1,5 +1,5 @@
import { getLogger } from '@stock-bot/logger';
import { sleep } from '@stock-bot/di';
import { getLogger } from '@stock-bot/logger';
const logger = getLogger('symbol-search-util');

204
apps/data-ingestion/test-ceo-operations.ts
View file

@ -1,101 +1,103 @@
#!/usr/bin/env bun
/**
* Test script for CEO handler operations
*/
import { initializeServiceConfig } from '@stock-bot/config';
import { createServiceContainer, initializeServices } from '@stock-bot/di';
import { getLogger } from '@stock-bot/logger';
const logger = getLogger('test-ceo-operations');
async function testCeoOperations() {
logger.info('Testing CEO handler operations...');
try {
// Initialize config
const config = initializeServiceConfig();
// Create Awilix container
const awilixConfig = {
redis: {
host: config.database.dragonfly.host,
port: config.database.dragonfly.port,
db: config.database.dragonfly.db,
},
mongodb: {
uri: config.database.mongodb.uri,
database: config.database.mongodb.database,
},
postgres: {
host: config.database.postgres.host,
port: config.database.postgres.port,
database: config.database.postgres.database,
user: config.database.postgres.user,
password: config.database.postgres.password,
},
questdb: {
enabled: false,
host: config.database.questdb.host,
httpPort: config.database.questdb.httpPort,
pgPort: config.database.questdb.pgPort,
influxPort: config.database.questdb.ilpPort,
database: config.database.questdb.database,
},
};
const container = createServiceContainer(awilixConfig);
await initializeServices(container);
const serviceContainer = container.resolve('serviceContainer');
// Import and create CEO handler
const { CeoHandler } = await import('./src/handlers/ceo/ceo.handler');
const ceoHandler = new CeoHandler(serviceContainer);
// Test 1: Check if there are any CEO symbols in the database
logger.info('Checking for existing CEO symbols...');
const collection = serviceContainer.mongodb.collection('ceoSymbols');
const count = await collection.countDocuments();
logger.info(`Found ${count} CEO symbols in database`);
if (count > 0) {
// Test 2: Run process-unique-symbols operation
logger.info('Testing process-unique-symbols operation...');
const result = await ceoHandler.updateUniqueSymbols(undefined, {});
logger.info('Process unique symbols result:', result);
// Test 3: Test individual symbol processing
logger.info('Testing process-individual-symbol operation...');
const sampleSymbol = await collection.findOne({});
if (sampleSymbol) {
const individualResult = await ceoHandler.processIndividualSymbol({
ceoId: sampleSymbol.ceoId,
symbol: sampleSymbol.symbol,
exchange: sampleSymbol.exchange,
name: sampleSymbol.name,
}, {});
logger.info('Process individual symbol result:', individualResult);
}
} else {
logger.warn('No CEO symbols found. Run the service to populate data first.');
}
// Clean up
await serviceContainer.mongodb.disconnect();
await serviceContainer.postgres.disconnect();
if (serviceContainer.cache) {
await serviceContainer.cache.disconnect();
}
logger.info('Test completed successfully!');
process.exit(0);
} catch (error) {
logger.error('Test failed:', error);
process.exit(1);
}
}
// Run the test
testCeoOperations();
#!/usr/bin/env bun
/**
* Test script for CEO handler operations
*/
import { initializeServiceConfig } from '@stock-bot/config';
import { createServiceContainer, initializeServices } from '@stock-bot/di';
import { getLogger } from '@stock-bot/logger';
const logger = getLogger('test-ceo-operations');
async function testCeoOperations() {
logger.info('Testing CEO handler operations...');
try {
// Initialize config
const config = initializeServiceConfig();
// Create Awilix container
const awilixConfig = {
redis: {
host: config.database.dragonfly.host,
port: config.database.dragonfly.port,
db: config.database.dragonfly.db,
},
mongodb: {
uri: config.database.mongodb.uri,
database: config.database.mongodb.database,
},
postgres: {
host: config.database.postgres.host,
port: config.database.postgres.port,
database: config.database.postgres.database,
user: config.database.postgres.user,
password: config.database.postgres.password,
},
questdb: {
enabled: false,
host: config.database.questdb.host,
httpPort: config.database.questdb.httpPort,
pgPort: config.database.questdb.pgPort,
influxPort: config.database.questdb.ilpPort,
database: config.database.questdb.database,
},
};
const container = createServiceContainer(awilixConfig);
await initializeServices(container);
const serviceContainer = container.resolve('serviceContainer');
// Import and create CEO handler
const { CeoHandler } = await import('./src/handlers/ceo/ceo.handler');
const ceoHandler = new CeoHandler(serviceContainer);
// Test 1: Check if there are any CEO symbols in the database
logger.info('Checking for existing CEO symbols...');
const collection = serviceContainer.mongodb.collection('ceoSymbols');
const count = await collection.countDocuments();
logger.info(`Found ${count} CEO symbols in database`);
if (count > 0) {
// Test 2: Run process-unique-symbols operation
logger.info('Testing process-unique-symbols operation...');
const result = await ceoHandler.updateUniqueSymbols(undefined, {});
logger.info('Process unique symbols result:', result);
// Test 3: Test individual symbol processing
logger.info('Testing process-individual-symbol operation...');
const sampleSymbol = await collection.findOne({});
if (sampleSymbol) {
const individualResult = await ceoHandler.processIndividualSymbol(
{
ceoId: sampleSymbol.ceoId,
symbol: sampleSymbol.symbol,
exchange: sampleSymbol.exchange,
name: sampleSymbol.name,
},
{}
);
logger.info('Process individual symbol result:', individualResult);
}
} else {
logger.warn('No CEO symbols found. Run the service to populate data first.');
}
// Clean up
await serviceContainer.mongodb.disconnect();
await serviceContainer.postgres.disconnect();
if (serviceContainer.cache) {
await serviceContainer.cache.disconnect();
}
logger.info('Test completed successfully!');
process.exit(0);
} catch (error) {
logger.error('Test failed:', error);
process.exit(1);
}
}
// Run the test
testCeoOperations();