di-refactor coming along
This commit is contained in:
parent
7d9044ab29
commit
60ada5f6a3
20 changed files with 582 additions and 335 deletions
|
|
@ -1,22 +1,31 @@
|
|||
/**
|
||||
* Data Pipeline Service with Dependency Injection
|
||||
* Uses Awilix container for managing database connections and services
|
||||
*/
|
||||
|
||||
// Framework imports
|
||||
import { Hono } from 'hono';
|
||||
import { cors } from 'hono/cors';
|
||||
import { initializeServiceConfig } from '@stock-bot/config';
|
||||
|
||||
// Library imports
|
||||
import {
|
||||
createServiceContainer,
|
||||
initializeServices as initializeAwilixServices,
|
||||
type ServiceContainer
|
||||
} from '@stock-bot/di';
|
||||
import { getLogger, setLoggerConfig, shutdownLoggers } from '@stock-bot/logger';
|
||||
import { MongoDBClient } from '@stock-bot/mongodb';
|
||||
import { PostgreSQLClient } from '@stock-bot/postgres';
|
||||
import { QueueManager, type QueueManagerConfig } from '@stock-bot/queue';
|
||||
import { Shutdown } from '@stock-bot/shutdown';
|
||||
import { setMongoDBClient, setPostgreSQLClient } from './clients';
|
||||
import { handlerRegistry } from '@stock-bot/types';
|
||||
|
||||
// Local imports
|
||||
import { enhancedSyncRoutes, healthRoutes, statsRoutes, syncRoutes } from './routes';
|
||||
import { createRoutes } from './routes/create-routes';
|
||||
import { setupServiceContainer } from './container-setup';
|
||||
import { initializeAllHandlers } from './handlers';
|
||||
|
||||
const config = initializeServiceConfig();
|
||||
console.log('Data Sync Service Configuration:', JSON.stringify(config, null, 2));
|
||||
console.log('Data Pipeline Service Configuration:', JSON.stringify(config, null, 2));
|
||||
const serviceConfig = config.service;
|
||||
const databaseConfig = config.database;
|
||||
const queueConfig = config.queue;
|
||||
|
||||
if (config.log) {
|
||||
setLoggerConfig({
|
||||
|
|
@ -31,129 +40,91 @@ if (config.log) {
|
|||
// Create logger AFTER config is set
|
||||
const logger = getLogger('data-pipeline');
|
||||
|
||||
const app = new Hono();
|
||||
|
||||
// Add CORS middleware
|
||||
app.use(
|
||||
'*',
|
||||
cors({
|
||||
origin: '*',
|
||||
allowMethods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'],
|
||||
allowHeaders: ['Content-Type', 'Authorization'],
|
||||
credentials: false,
|
||||
})
|
||||
);
|
||||
const PORT = serviceConfig.port;
|
||||
let server: ReturnType<typeof Bun.serve> | null = null;
|
||||
let mongoClient: MongoDBClient | null = null;
|
||||
let postgresClient: PostgreSQLClient | null = null;
|
||||
let queueManager: QueueManager | null = null;
|
||||
let container: ServiceContainer | null = null;
|
||||
let app: Hono | null = null;
|
||||
|
||||
// Initialize shutdown manager
|
||||
const shutdown = Shutdown.getInstance({ timeout: 15000 });
|
||||
|
||||
// Mount routes
|
||||
app.route('/health', healthRoutes);
|
||||
app.route('/sync', syncRoutes);
|
||||
app.route('/sync', enhancedSyncRoutes);
|
||||
app.route('/sync/stats', statsRoutes);
|
||||
|
||||
// Initialize services
|
||||
// Initialize services with DI pattern
|
||||
async function initializeServices() {
|
||||
logger.info('Initializing data sync service...');
|
||||
logger.info('Initializing data pipeline service with DI...');
|
||||
|
||||
try {
|
||||
// Initialize MongoDB client
|
||||
logger.debug('Connecting to MongoDB...');
|
||||
const mongoConfig = databaseConfig.mongodb;
|
||||
mongoClient = new MongoDBClient(
|
||||
{
|
||||
uri: mongoConfig.uri,
|
||||
database: mongoConfig.database,
|
||||
host: mongoConfig.host || 'localhost',
|
||||
port: mongoConfig.port || 27017,
|
||||
timeouts: {
|
||||
connectTimeout: 30000,
|
||||
socketTimeout: 30000,
|
||||
serverSelectionTimeout: 5000,
|
||||
},
|
||||
// Create Awilix container with proper config structure
|
||||
logger.debug('Creating Awilix DI container...');
|
||||
const awilixConfig = {
|
||||
redis: {
|
||||
host: config.database.dragonfly.host,
|
||||
port: config.database.dragonfly.port,
|
||||
db: config.database.dragonfly.db,
|
||||
},
|
||||
logger
|
||||
);
|
||||
await mongoClient.connect();
|
||||
setMongoDBClient(mongoClient);
|
||||
logger.info('MongoDB connected');
|
||||
|
||||
// Initialize PostgreSQL client
|
||||
logger.debug('Connecting to PostgreSQL...');
|
||||
const pgConfig = databaseConfig.postgres;
|
||||
postgresClient = new PostgreSQLClient(
|
||||
{
|
||||
host: pgConfig.host,
|
||||
port: pgConfig.port,
|
||||
database: pgConfig.database,
|
||||
username: pgConfig.user,
|
||||
password: pgConfig.password,
|
||||
poolSettings: {
|
||||
min: 2,
|
||||
max: pgConfig.poolSize || 10,
|
||||
idleTimeoutMillis: pgConfig.idleTimeout || 30000,
|
||||
},
|
||||
mongodb: {
|
||||
uri: config.database.mongodb.uri,
|
||||
database: config.database.mongodb.database,
|
||||
},
|
||||
logger
|
||||
);
|
||||
await postgresClient.connect();
|
||||
setPostgreSQLClient(postgresClient);
|
||||
logger.info('PostgreSQL connected');
|
||||
|
||||
// Initialize queue system (with delayed worker start)
|
||||
logger.debug('Initializing queue system...');
|
||||
const queueManagerConfig: QueueManagerConfig = {
|
||||
redis: queueConfig?.redis || {
|
||||
host: 'localhost',
|
||||
port: 6379,
|
||||
db: 1,
|
||||
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,
|
||||
},
|
||||
defaultQueueOptions: {
|
||||
defaultJobOptions: queueConfig?.defaultJobOptions || {
|
||||
attempts: 3,
|
||||
backoff: {
|
||||
type: 'exponential',
|
||||
delay: 1000,
|
||||
},
|
||||
removeOnComplete: 10,
|
||||
removeOnFail: 5,
|
||||
},
|
||||
workers: 2,
|
||||
concurrency: 1,
|
||||
enableMetrics: true,
|
||||
enableDLQ: true,
|
||||
questdb: {
|
||||
enabled: config.database.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,
|
||||
},
|
||||
enableScheduledJobs: true,
|
||||
delayWorkerStart: true, // Prevent workers from starting until all singletons are ready
|
||||
};
|
||||
|
||||
container = createServiceContainer(awilixConfig);
|
||||
await initializeAwilixServices(container);
|
||||
logger.info('Awilix container created and initialized');
|
||||
|
||||
// Setup service-specific configuration
|
||||
const serviceContainer = setupServiceContainer(config, container.resolve('serviceContainer'));
|
||||
|
||||
// Initialize migration helper for backward compatibility
|
||||
const { setContainerForMigration } = await import('./migration-helper');
|
||||
setContainerForMigration(serviceContainer);
|
||||
logger.info('Migration helper initialized for backward compatibility');
|
||||
|
||||
// Create app with routes
|
||||
app = new Hono();
|
||||
|
||||
// Add CORS middleware
|
||||
app.use(
|
||||
'*',
|
||||
cors({
|
||||
origin: '*',
|
||||
allowMethods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'],
|
||||
allowHeaders: ['Content-Type', 'Authorization'],
|
||||
credentials: false,
|
||||
})
|
||||
);
|
||||
|
||||
// Create and mount routes using the service container
|
||||
const routes = createRoutes(serviceContainer);
|
||||
app.route('/', routes);
|
||||
|
||||
queueManager = QueueManager.getOrInitialize(queueManagerConfig);
|
||||
logger.info('Queue system initialized');
|
||||
|
||||
// Initialize handlers (register handlers and scheduled jobs)
|
||||
logger.debug('Initializing sync handlers...');
|
||||
const { initializeExchangesHandler } = await import('./handlers/exchanges/exchanges.handler');
|
||||
const { initializeSymbolsHandler } = await import('./handlers/symbols/symbols.handler');
|
||||
|
||||
initializeExchangesHandler();
|
||||
initializeSymbolsHandler();
|
||||
|
||||
logger.info('Sync handlers initialized');
|
||||
// Initialize handlers with service container
|
||||
logger.debug('Initializing pipeline handlers with DI pattern...');
|
||||
await initializeAllHandlers(serviceContainer);
|
||||
logger.info('Pipeline handlers initialized with DI pattern');
|
||||
|
||||
// Create scheduled jobs from registered handlers
|
||||
logger.debug('Creating scheduled jobs from registered handlers...');
|
||||
const { handlerRegistry } = await import('@stock-bot/queue');
|
||||
const allHandlers = handlerRegistry.getAllHandlers();
|
||||
const allHandlers = handlerRegistry.getAllHandlersWithSchedule();
|
||||
|
||||
let totalScheduledJobs = 0;
|
||||
for (const [handlerName, config] of allHandlers) {
|
||||
if (config.scheduledJobs && config.scheduledJobs.length > 0) {
|
||||
const queueManager = container!.resolve('queueManager');
|
||||
const queue = queueManager.getQueue(handlerName);
|
||||
|
||||
for (const scheduledJob of config.scheduledJobs) {
|
||||
|
|
@ -161,7 +132,7 @@ async function initializeServices() {
|
|||
const jobData = {
|
||||
handler: handlerName,
|
||||
operation: scheduledJob.operation,
|
||||
payload: scheduledJob.payload || {},
|
||||
payload: scheduledJob.payload,
|
||||
};
|
||||
|
||||
// Build job options from scheduled job config
|
||||
|
|
@ -192,14 +163,22 @@ async function initializeServices() {
|
|||
}
|
||||
logger.info('Scheduled jobs created', { totalJobs: totalScheduledJobs });
|
||||
|
||||
// Now that all singletons are initialized and jobs are scheduled, start the workers
|
||||
// Start queue workers
|
||||
logger.debug('Starting queue workers...');
|
||||
queueManager.startAllWorkers();
|
||||
logger.info('Queue workers started');
|
||||
const queueManager = container.resolve('queueManager');
|
||||
if (queueManager) {
|
||||
queueManager.startAllWorkers();
|
||||
logger.info('Queue workers started');
|
||||
}
|
||||
|
||||
logger.info('All services initialized successfully');
|
||||
} catch (error) {
|
||||
logger.error('Failed to initialize services', { error });
|
||||
console.error('DETAILED ERROR:', error);
|
||||
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)
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
|
@ -208,13 +187,17 @@ async function initializeServices() {
|
|||
async function startServer() {
|
||||
await initializeServices();
|
||||
|
||||
if (!app) {
|
||||
throw new Error('App not initialized');
|
||||
}
|
||||
|
||||
server = Bun.serve({
|
||||
port: PORT,
|
||||
fetch: app.fetch,
|
||||
development: config.environment === 'development',
|
||||
});
|
||||
|
||||
logger.info(`Data Sync Service started on port ${PORT}`);
|
||||
logger.info(`Data pipeline service started on port ${PORT}`);
|
||||
}
|
||||
|
||||
// Register shutdown handlers with priorities
|
||||
|
|
@ -222,6 +205,7 @@ async function startServer() {
|
|||
shutdown.onShutdownHigh(async () => {
|
||||
logger.info('Shutting down queue system...');
|
||||
try {
|
||||
const queueManager = container?.resolve('queueManager');
|
||||
if (queueManager) {
|
||||
await queueManager.shutdown();
|
||||
}
|
||||
|
|
@ -244,21 +228,27 @@ shutdown.onShutdownHigh(async () => {
|
|||
}
|
||||
}, 'HTTP Server');
|
||||
|
||||
// Priority 2: Database connections (medium priority)
|
||||
// Priority 2: Services and connections (medium priority)
|
||||
shutdown.onShutdownMedium(async () => {
|
||||
logger.info('Disconnecting from databases...');
|
||||
logger.info('Disposing services and connections...');
|
||||
try {
|
||||
if (mongoClient) {
|
||||
await mongoClient.disconnect();
|
||||
if (container) {
|
||||
// Disconnect database clients
|
||||
const mongoClient = container.resolve('mongoClient');
|
||||
if (mongoClient?.disconnect) await mongoClient.disconnect();
|
||||
|
||||
const postgresClient = container.resolve('postgresClient');
|
||||
if (postgresClient?.disconnect) await postgresClient.disconnect();
|
||||
|
||||
const questdbClient = container.resolve('questdbClient');
|
||||
if (questdbClient?.disconnect) await questdbClient.disconnect();
|
||||
|
||||
logger.info('All services disposed successfully');
|
||||
}
|
||||
if (postgresClient) {
|
||||
await postgresClient.disconnect();
|
||||
}
|
||||
logger.info('Database connections closed');
|
||||
} catch (error) {
|
||||
logger.error('Error closing database connections', { error });
|
||||
logger.error('Error disposing services', { error });
|
||||
}
|
||||
}, 'Databases');
|
||||
}, 'Services');
|
||||
|
||||
// Priority 3: Logger shutdown (lowest priority - runs last)
|
||||
shutdown.onShutdownLow(async () => {
|
||||
|
|
@ -273,8 +263,8 @@ shutdown.onShutdownLow(async () => {
|
|||
|
||||
// Start the service
|
||||
startServer().catch(error => {
|
||||
logger.fatal('Failed to start data sync service', { error });
|
||||
logger.fatal('Failed to start data pipeline service', { error });
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
logger.info('Data sync service startup initiated');
|
||||
logger.info('Data pipeline service startup initiated with DI pattern');
|
||||
Loading…
Add table
Add a link
Reference in a new issue