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

unified config

Browse source
This commit is contained in:
Boki 2025-06-23 11:16:34 -04:00
parent e7c0fe2798
commit 3877902ff4
13 changed files with 856 additions and 476 deletions
Download patch file Download diff file Expand all files Collapse all files

4
apps/stock/config/src/config-instance.ts
View file

@ -20,12 +20,14 @@ export function initializeStockConfig(serviceName?: 'dataIngestion' | 'dataPipel
// If a service name is provided, override the service port
if (serviceName && config.services?.[serviceName]) {
const kebabName = serviceName.replace(/([A-Z])/g, '-$1').toLowerCase().replace(/^-/, '');
return {
...config,
service: {
...config.service,
port: config.services[serviceName].port,
name: serviceName.replace(/([A-Z])/g, '-$1').toLowerCase() // Convert camelCase to kebab-case
name: serviceName, // Keep original for backward compatibility
serviceName: kebabName // Standard kebab-case name
}
};
}

11
apps/stock/web-api/src/index.ts
View file

@ -47,17 +47,6 @@ const app = new ServiceApplication(
},
{
// Custom lifecycle hooks
onContainerReady: (container) => {
// Override queue configuration to disable workers
const config = container.cradle.config;
if (config.queue) {
config.queue.workers = 0;
config.queue.concurrency = 0;
config.queue.enableScheduledJobs = false;
config.queue.delayWorkerStart = true;
}
return container;
},
onStarted: (port) => {
const logger = getLogger('web-api');
logger.info('Web API service startup initiated with ServiceApplication framework');

133
docs/configuration-standardization.md Normal file
View file

@ -0,0 +1,133 @@
# Configuration Standardization
## Overview
The Stock Bot system now uses a unified configuration approach that standardizes how services receive and use configuration. This eliminates the previous confusion between `StockBotAppConfig` and `AppConfig`, providing a single source of truth for all configuration needs.
## Key Changes
### 1. Unified Configuration Schema
The new `UnifiedAppConfig` schema:
- Provides both nested (backward compatible) and flat (DI-friendly) database configurations
- Automatically standardizes service names to kebab-case
- Handles field name mappings (e.g., `ilpPort` → `influxPort`)
- Ensures all required fields are present for DI system
### 2. Service Name Standardization
All service names are now standardized to kebab-case:
- `dataIngestion` → `data-ingestion`
- `dataPipeline` → `data-pipeline`
- `webApi` → `web-api`
This happens automatically in:
- `initializeStockConfig()` when passing service name
- `ServiceApplication` constructor
- `toUnifiedConfig()` transformation
### 3. Single Configuration Object
Services now use a single configuration object (`this.config`) that contains:
- All service-specific settings
- Database configurations (both nested and flat)
- Service metadata including standardized name
- All settings required by the DI system
## Migration Guide
### For Service Implementations
Before:
```typescript
const app = new ServiceApplication(
config,
{
serviceName: 'web-api',
// other options
}
);
// In container factory
const configWithService = {
...this.config,
service: { name: this.serviceConfig.serviceName }
};
```
After:
```typescript
const app = new ServiceApplication(
config, // Config already has service.serviceName
{
serviceName: 'web-api', // Still needed for logger
// other options
}
);
// In container factory
// No manual service name addition needed
this.container = await containerFactory(this.config);
```
### For DI Container Usage
Before:
```typescript
const serviceName = config.service?.name || 'unknown';
// Had to handle different naming conventions
```
After:
```typescript
const serviceName = config.service?.serviceName || config.service?.name || 'unknown';
// Standardized kebab-case name is always available
```
### For Configuration Files
The configuration structure remains the same, but the system now ensures:
- Service names are standardized automatically
- Database configs are available in both formats
- All required fields are properly mapped
## Benefits
1. **Simplicity**: One configuration object with all necessary information
2. **Consistency**: Standardized service naming across the system
3. **Type Safety**: Unified schema provides better TypeScript support
4. **Backward Compatibility**: Old configuration formats still work
5. **Reduced Complexity**: No more manual config transformations
## Technical Details
### UnifiedAppConfig Schema
```typescript
export const unifiedAppSchema = baseAppSchema.extend({
// Flat database configs for DI system
redis: dragonflyConfigSchema.optional(),
mongodb: mongodbConfigSchema.optional(),
postgres: postgresConfigSchema.optional(),
questdb: questdbConfigSchema.optional(),
}).transform((data) => {
// Auto-standardize service name
// Sync nested and flat configs
// Handle field mappings
});
```
### Service Registry
The `SERVICE_REGISTRY` now includes aliases for different naming conventions:
```typescript
'web-api': { db: 3, ... },
'webApi': { db: 3, ... }, // Alias for backward compatibility
```
## Future Improvements
1. Remove service name aliases after full migration
2. Deprecate old configuration formats
3. Add configuration validation at startup
4. Provide migration tooling for existing services

155
libs/core/config/src/schemas/__tests__/unified-app.test.ts Normal file
View file

@ -0,0 +1,155 @@
import { describe, expect, it } from 'bun:test';
import { unifiedAppSchema, toUnifiedConfig, getStandardServiceName } from '../unified-app.schema';
describe('UnifiedAppConfig', () => {
describe('getStandardServiceName', () => {
it('should convert camelCase to kebab-case', () => {
expect(getStandardServiceName('dataIngestion')).toBe('data-ingestion');
expect(getStandardServiceName('dataPipeline')).toBe('data-pipeline');
expect(getStandardServiceName('webApi')).toBe('web-api');
});
it('should handle already kebab-case names', () => {
expect(getStandardServiceName('data-ingestion')).toBe('data-ingestion');
expect(getStandardServiceName('web-api')).toBe('web-api');
});
it('should handle single word names', () => {
expect(getStandardServiceName('api')).toBe('api');
expect(getStandardServiceName('worker')).toBe('worker');
});
});
describe('unifiedAppSchema transform', () => {
it('should set serviceName from name if not provided', () => {
const config = {
name: 'test-app',
version: '1.0.0',
service: {
name: 'webApi',
port: 3000,
},
log: { level: 'info' },
};
const result = unifiedAppSchema.parse(config);
expect(result.service.serviceName).toBe('web-api');
});
it('should keep existing serviceName if provided', () => {
const config = {
name: 'test-app',
version: '1.0.0',
service: {
name: 'webApi',
serviceName: 'custom-name',
port: 3000,
},
log: { level: 'info' },
};
const result = unifiedAppSchema.parse(config);
expect(result.service.serviceName).toBe('custom-name');
});
it('should sync nested and flat database configs', () => {
const config = {
name: 'test-app',
version: '1.0.0',
service: { name: 'test', port: 3000 },
log: { level: 'info' },
database: {
postgres: {
host: 'localhost',
port: 5432,
database: 'test',
user: 'user',
password: 'pass',
},
mongodb: {
uri: 'mongodb://localhost:27017',
database: 'test',
},
},
};
const result = unifiedAppSchema.parse(config);
// Should have both nested and flat structure
expect(result.postgres).toBeDefined();
expect(result.mongodb).toBeDefined();
expect(result.database?.postgres).toBeDefined();
expect(result.database?.mongodb).toBeDefined();
// Values should match
expect(result.postgres?.host).toBe('localhost');
expect(result.postgres?.port).toBe(5432);
expect(result.mongodb?.uri).toBe('mongodb://localhost:27017');
});
it('should handle questdb ilpPort to influxPort mapping', () => {
const config = {
name: 'test-app',
version: '1.0.0',
service: { name: 'test', port: 3000 },
log: { level: 'info' },
database: {
questdb: {
host: 'localhost',
ilpPort: 9009,
httpPort: 9000,
pgPort: 8812,
database: 'questdb',
},
},
};
const result = unifiedAppSchema.parse(config);
expect(result.questdb).toBeDefined();
expect((result.questdb as any).influxPort).toBe(9009);
});
});
describe('toUnifiedConfig', () => {
it('should convert StockBotAppConfig to UnifiedAppConfig', () => {
const stockBotConfig = {
name: 'stock-bot',
version: '1.0.0',
environment: 'development',
service: {
name: 'dataIngestion',
port: 3001,
host: '0.0.0.0',
},
log: {
level: 'info',
format: 'json',
},
database: {
postgres: {
enabled: true,
host: 'localhost',
port: 5432,
database: 'stock',
user: 'user',
password: 'pass',
},
dragonfly: {
enabled: true,
host: 'localhost',
port: 6379,
db: 0,
},
},
};
const unified = toUnifiedConfig(stockBotConfig);
expect(unified.service.serviceName).toBe('data-ingestion');
expect(unified.redis).toBeDefined();
expect(unified.redis?.host).toBe('localhost');
expect(unified.postgres).toBeDefined();
expect(unified.postgres?.host).toBe('localhost');
});
});
});

4
libs/core/config/src/schemas/index.ts
View file

@ -12,6 +12,10 @@ export * from './provider.schema';
export { baseAppSchema } from './base-app.schema';
export type { BaseAppConfig } from './base-app.schema';
// Export unified schema for standardized configuration
export { unifiedAppSchema, toUnifiedConfig, getStandardServiceName } from './unified-app.schema';
export type { UnifiedAppConfig } from './unified-app.schema';
// Keep AppConfig for backward compatibility (deprecated)
// @deprecated Use baseAppSchema and extend it for your specific app
import { z } from 'zod';

6
libs/core/config/src/schemas/service.schema.ts
View file

@ -3,6 +3,7 @@ import { z } from 'zod';
// Common service configuration
export const serviceConfigSchema = z.object({
name: z.string(),
serviceName: z.string().optional(), // Standard service name (kebab-case)
port: z.number().min(1).max(65535),
host: z.string().default('0.0.0.0'),
healthCheckPath: z.string().default('/health'),
@ -96,6 +97,11 @@ export const browserConfigSchema = z.object({
// Proxy manager configuration
export const proxyConfigSchema = z.object({
enabled: z.boolean().default(false),
cachePrefix: z.string().default('proxy:'),
ttl: z.number().default(3600),
webshare: z.object({
apiKey: z.string(),
apiUrl: z.string().default('https://proxy.webshare.io/api/v2/'),
}).optional(),
});

76
libs/core/config/src/schemas/unified-app.schema.ts Normal file
View file

@ -0,0 +1,76 @@
import { z } from 'zod';
import { baseAppSchema } from './base-app.schema';
import {
postgresConfigSchema,
mongodbConfigSchema,
questdbConfigSchema,
dragonflyConfigSchema
} from './database.schema';
/**
* Unified application configuration schema that provides both nested and flat access
* to database configurations for backward compatibility while maintaining a clean structure
*/
export const unifiedAppSchema = baseAppSchema.extend({
// Flat database configs for DI system (these take precedence)
redis: dragonflyConfigSchema.optional(),
mongodb: mongodbConfigSchema.optional(),
postgres: postgresConfigSchema.optional(),
questdb: questdbConfigSchema.optional(),
}).transform((data) => {
// Ensure service.serviceName is set from service.name if not provided
if (data.service && !data.service.serviceName) {
data.service.serviceName = data.service.name.replace(/([A-Z])/g, '-$1').toLowerCase().replace(/^-/, '');
}
// If flat configs exist, ensure they're also in the nested database object
if (data.redis || data.mongodb || data.postgres || data.questdb) {
data.database = {
...data.database,
dragonfly: data.redis || data.database?.dragonfly,
mongodb: data.mongodb || data.database?.mongodb,
postgres: data.postgres || data.database?.postgres,
questdb: data.questdb || data.database?.questdb,
};
}
// If nested configs exist but flat ones don't, copy them to flat structure
if (data.database) {
if (data.database.dragonfly && !data.redis) {
data.redis = data.database.dragonfly;
}
if (data.database.mongodb && !data.mongodb) {
data.mongodb = data.database.mongodb;
}
if (data.database.postgres && !data.postgres) {
data.postgres = data.database.postgres;
}
if (data.database.questdb && !data.questdb) {
// Handle the ilpPort -> influxPort mapping for DI system
const questdbConfig = { ...data.database.questdb };
if ('ilpPort' in questdbConfig && !('influxPort' in questdbConfig)) {
(questdbConfig as any).influxPort = questdbConfig.ilpPort;
}
data.questdb = questdbConfig;
}
}
return data;
});
export type UnifiedAppConfig = z.infer<typeof unifiedAppSchema>;
/**
* Helper to convert StockBotAppConfig to UnifiedAppConfig
*/
export function toUnifiedConfig(config: any): UnifiedAppConfig {
return unifiedAppSchema.parse(config);
}
/**
* Helper to get standardized service name
*/
export function getStandardServiceName(serviceName: string): string {
// Convert camelCase to kebab-case
return serviceName.replace(/([A-Z])/g, '-$1').toLowerCase().replace(/^-/, '');
}

2
libs/core/di/src/config/schemas/index.ts
View file

@ -15,6 +15,8 @@ export const appConfigSchema = z.object({
queue: queueConfigSchema.optional(),
service: z.object({
name: z.string(),
serviceName: z.string().optional(), // Standard kebab-case service name
port: z.number().optional(),
}).optional(),
});

97
libs/core/di/src/container/builder.ts
View file

@ -1,6 +1,7 @@
import { createContainer, InjectionMode, asFunction, type AwilixContainer } from 'awilix';
import type { AppConfig as StockBotAppConfig } from '@stock-bot/config';
import type { AppConfig as StockBotAppConfig, UnifiedAppConfig } from '@stock-bot/config';
import { appConfigSchema, type AppConfig } from '../config/schemas';
import { toUnifiedConfig } from '@stock-bot/config';
import {
registerCoreServices,
registerCacheServices,
@ -12,6 +13,7 @@ import type { ServiceDefinitions, ContainerBuildOptions } from './types';
export class ServiceContainerBuilder {
private config: Partial<AppConfig> = {};
private unifiedConfig: UnifiedAppConfig | null = null;
private options: ContainerBuildOptions = {
enableCache: true,
enableQueue: true,
@ -24,8 +26,10 @@ export class ServiceContainerBuilder {
initializationTimeout: 30000,
};
withConfig(config: AppConfig | StockBotAppConfig): this {
this.config = this.transformStockBotConfig(config);
withConfig(config: AppConfig | StockBotAppConfig | UnifiedAppConfig): this {
// Convert to unified config format
this.unifiedConfig = toUnifiedConfig(config);
this.config = this.transformStockBotConfig(this.unifiedConfig);
return this;
}
@ -72,6 +76,19 @@ export class ServiceContainerBuilder {
}
private applyServiceOptions(config: Partial<AppConfig>): AppConfig {
// Ensure questdb config has the right field names for DI
const questdbConfig = config.questdb ? {
...config.questdb,
influxPort: (config.questdb as any).influxPort || (config.questdb as any).ilpPort || 9009,
} : {
enabled: true,
host: 'localhost',
httpPort: 9000,
pgPort: 8812,
influxPort: 9009,
database: 'questdb',
};
return {
redis: config.redis || {
enabled: this.options.enableCache ?? true,
@ -92,14 +109,7 @@ export class ServiceContainerBuilder {
user: 'postgres',
password: 'postgres',
},
questdb: this.options.enableQuestDB ? (config.questdb || {
enabled: true,
host: 'localhost',
httpPort: 9000,
pgPort: 8812,
influxPort: 9009,
database: 'questdb',
}) : undefined,
questdb: this.options.enableQuestDB ? questdbConfig : undefined,
proxy: this.options.enableProxy ? (config.proxy || { enabled: false, cachePrefix: 'proxy:', ttl: 3600 }) : undefined,
browser: this.options.enableBrowser ? (config.browser || { headless: true, timeout: 30000 }) : undefined,
queue: this.options.enableQueue ? (config.queue || {
@ -115,6 +125,7 @@ export class ServiceContainerBuilder {
removeOnFail: 50,
}
}) : undefined,
service: config.service,
};
}
@ -143,53 +154,27 @@ export class ServiceContainerBuilder {
});
}
private transformStockBotConfig(config: AppConfig | StockBotAppConfig): Partial<AppConfig> {
// If it's already in the new format (has redis AND postgres at top level), return as is
if ('redis' in config && 'postgres' in config && 'mongodb' in config) {
return config as AppConfig;
}
private transformStockBotConfig(config: UnifiedAppConfig): Partial<AppConfig> {
// Unified config already has flat structure, just extract what we need
// Handle questdb field name mapping
const questdb = config.questdb ? {
enabled: config.questdb.enabled || true,
host: config.questdb.host || 'localhost',
httpPort: config.questdb.httpPort || 9000,
pgPort: config.questdb.pgPort || 8812,
influxPort: (config.questdb as any).influxPort || (config.questdb as any).ilpPort || 9009,
database: config.questdb.database || 'questdb',
} : undefined;
// Transform from StockBotAppConfig format
const stockBotConfig = config as StockBotAppConfig;
return {
redis: stockBotConfig.database?.dragonfly ? {
enabled: true,
host: stockBotConfig.database.dragonfly.host || 'localhost',
port: stockBotConfig.database.dragonfly.port || 6379,
password: stockBotConfig.database.dragonfly.password,
db: stockBotConfig.database.dragonfly.db || 0,
} : undefined,
mongodb: stockBotConfig.database?.mongodb ? {
enabled: stockBotConfig.database.mongodb.enabled ?? true,
uri: stockBotConfig.database.mongodb.uri,
database: stockBotConfig.database.mongodb.database,
} : undefined,
postgres: stockBotConfig.database?.postgres ? {
enabled: stockBotConfig.database.postgres.enabled ?? true,
host: stockBotConfig.database.postgres.host,
port: stockBotConfig.database.postgres.port,
database: stockBotConfig.database.postgres.database,
user: stockBotConfig.database.postgres.user,
password: stockBotConfig.database.postgres.password,
} : undefined,
questdb: stockBotConfig.database?.questdb ? {
enabled: true,
host: stockBotConfig.database.questdb.host || 'localhost',
httpPort: stockBotConfig.database.questdb.httpPort || 9000,
pgPort: stockBotConfig.database.questdb.pgPort || 8812,
influxPort: stockBotConfig.database.questdb.ilpPort || 9009,
database: stockBotConfig.database.questdb.database || 'questdb',
} : undefined,
queue: stockBotConfig.queue,
browser: stockBotConfig.browser,
proxy: stockBotConfig.proxy ? {
...{
enabled: false,
cachePrefix: 'proxy:',
ttl: 3600,
},
...stockBotConfig.proxy
} : undefined,
redis: config.redis,
mongodb: config.mongodb,
postgres: config.postgres,
questdb,
queue: config.queue,
browser: config.browser,
proxy: config.proxy,
service: config.service,
};
}
}

5
libs/core/di/src/registrations/cache.registration.ts
View file

@ -11,7 +11,8 @@ export function registerCacheServices(
container.register({
cache: asFunction(() => {
const { createServiceCache } = require('@stock-bot/queue');
const serviceName = config.service?.name || 'unknown';
// Get standardized service name from config
const serviceName = config.service?.serviceName || config.service?.name || 'unknown';
// Create service-specific cache that uses the service's Redis DB
return createServiceCache(serviceName, {
@ -25,7 +26,7 @@ export function registerCacheServices(
// Also provide global cache for shared data
globalCache: asFunction(() => {
const { createServiceCache } = require('@stock-bot/queue');
const serviceName = config.service?.name || 'unknown';
const serviceName = config.service?.serviceName || config.service?.name || 'unknown';
return createServiceCache(serviceName, {
host: config.redis.host,

2
libs/core/di/src/registrations/service.registration.ts
View file

@ -53,7 +53,7 @@ export function registerApplicationServices(
queueManager: asFunction(({ logger }) => {
const { SmartQueueManager } = require('@stock-bot/queue');
const queueConfig = {
serviceName: config.service?.name || 'unknown',
serviceName: config.service?.serviceName || config.service?.name || 'unknown',
redis: {
host: config.redis.host,
port: config.redis.port,

817
libs/core/di/src/service-application.ts
View file

@ -1,405 +1,414 @@
/**
* ServiceApplication - Common service initialization and lifecycle management
* Encapsulates common patterns for Hono-based microservices
*/
import { Hono } from 'hono';
import { cors } from 'hono/cors';
import { getLogger, setLoggerConfig, shutdownLoggers, type Logger } from '@stock-bot/logger';
import { Shutdown } from '@stock-bot/shutdown';
import type { AppConfig as StockBotAppConfig } from '@stock-bot/config';
import type { IServiceContainer } from '@stock-bot/handlers';
import type { ServiceContainer } from './awilix-container';
/**
* Configuration for ServiceApplication
*/
export interface ServiceApplicationConfig {
/** Service name for logging and identification */
serviceName: string;
/** CORS configuration - if not provided, uses permissive defaults */
corsConfig?: Parameters<typeof cors>[0];
/** Whether to enable handler initialization */
enableHandlers?: boolean;
/** Whether to enable scheduled job creation */
enableScheduledJobs?: boolean;
/** Custom shutdown timeout in milliseconds */
shutdownTimeout?: number;
/** Service metadata for info endpoint */
serviceMetadata?: {
version?: string;
description?: string;
endpoints?: Record<string, string>;
};
/** Whether to add a basic info endpoint at root */
addInfoEndpoint?: boolean;
}
/**
* Lifecycle hooks for service customization
*/
export interface ServiceLifecycleHooks {
/** Called after container is created but before routes */
onContainerReady?: (container: IServiceContainer) => Promise<void> | void;
/** Called after app is created but before routes are mounted */
onAppReady?: (app: Hono, container: IServiceContainer) => Promise<void> | void;
/** Called after routes are mounted but before server starts */
onBeforeStart?: (app: Hono, container: IServiceContainer) => Promise<void> | void;
/** Called after successful server startup */
onStarted?: (port: number) => Promise<void> | void;
/** Called during shutdown before cleanup */
onBeforeShutdown?: () => Promise<void> | void;
}
/**
* ServiceApplication - Manages the complete lifecycle of a microservice
*/
export class ServiceApplication {
private config: StockBotAppConfig;
private serviceConfig: ServiceApplicationConfig;
private hooks: ServiceLifecycleHooks;
private logger: Logger;
private container: ServiceContainer | null = null;
private serviceContainer: IServiceContainer | null = null;
private app: Hono | null = null;
private server: ReturnType<typeof Bun.serve> | null = null;
private shutdown: Shutdown;
constructor(
config: StockBotAppConfig,
serviceConfig: ServiceApplicationConfig,
hooks: ServiceLifecycleHooks = {}
) {
this.config = config;
this.serviceConfig = {
shutdownTimeout: 15000,
enableHandlers: false,
enableScheduledJobs: false,
addInfoEndpoint: true,
...serviceConfig,
};
this.hooks = hooks;
// Initialize logger configuration
this.configureLogger();
this.logger = getLogger(this.serviceConfig.serviceName);
// Initialize shutdown manager
this.shutdown = Shutdown.getInstance({
timeout: this.serviceConfig.shutdownTimeout
});
}
/**
* Configure logger based on application config
*/
private configureLogger(): void {
if (this.config.log) {
setLoggerConfig({
logLevel: this.config.log.level,
logConsole: true,
logFile: false,
environment: this.config.environment,
hideObject: this.config.log.hideObject,
});
}
}
/**
* Create and configure Hono application with CORS
*/
private createApp(): Hono {
const app = new Hono();
// Add CORS middleware with service-specific or default configuration
const corsConfig = this.serviceConfig.corsConfig || {
origin: '*',
allowMethods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'],
allowHeaders: ['Content-Type', 'Authorization'],
credentials: false,
};
app.use('*', cors(corsConfig));
// Add basic info endpoint if enabled
if (this.serviceConfig.addInfoEndpoint) {
const metadata = this.serviceConfig.serviceMetadata || {};
app.get('/', c => {
return c.json({
name: this.serviceConfig.serviceName,
version: metadata.version || '1.0.0',
description: metadata.description,
status: 'running',
timestamp: new Date().toISOString(),
endpoints: metadata.endpoints || {},
});
});
}
return app;
}
/**
* Register graceful shutdown handlers
*/
private registerShutdownHandlers(): void {
// Priority 1: Queue system (highest priority)
if (this.serviceConfig.enableScheduledJobs) {
this.shutdown.onShutdownHigh(async () => {
this.logger.info('Shutting down queue system...');
try {
const queueManager = this.container?.resolve('queueManager');
if (queueManager) {
await queueManager.shutdown();
}
this.logger.info('Queue system shut down');
} catch (error) {
this.logger.error('Error shutting down queue system', { error });
}
}, 'Queue System');
}
// Priority 1: HTTP Server (high priority)
this.shutdown.onShutdownHigh(async () => {
if (this.server) {
this.logger.info('Stopping HTTP server...');
try {
this.server.stop();
this.logger.info('HTTP server stopped');
} catch (error) {
this.logger.error('Error stopping HTTP server', { error });
}
}
}, 'HTTP Server');
// Custom shutdown hook
if (this.hooks.onBeforeShutdown) {
this.shutdown.onShutdownHigh(async () => {
try {
await this.hooks.onBeforeShutdown!();
} catch (error) {
this.logger.error('Error in custom shutdown hook', { error });
}
}, 'Custom Shutdown');
}
// Priority 2: Services and connections (medium priority)
this.shutdown.onShutdownMedium(async () => {
this.logger.info('Disposing services and connections...');
try {
if (this.container) {
// Disconnect database clients
const mongoClient = this.container.resolve('mongoClient');
if (mongoClient?.disconnect) {
await mongoClient.disconnect();
}
const postgresClient = this.container.resolve('postgresClient');
if (postgresClient?.disconnect) {
await postgresClient.disconnect();
}
const questdbClient = this.container.resolve('questdbClient');
if (questdbClient?.disconnect) {
await questdbClient.disconnect();
}
this.logger.info('All services disposed successfully');
}
} catch (error) {
this.logger.error('Error disposing services', { error });
}
}, 'Services');
// Priority 3: Logger shutdown (lowest priority - runs last)
this.shutdown.onShutdownLow(async () => {
try {
this.logger.info('Shutting down loggers...');
await shutdownLoggers();
// Don't log after shutdown
} catch {
// Silently ignore logger shutdown errors
}
}, 'Loggers');
}
/**
* Start the service with full initialization
*/
async start(
containerFactory: (config: StockBotAppConfig) => Promise<ServiceContainer>,
routeFactory: (container: IServiceContainer) => Hono,
handlerInitializer?: (container: IServiceContainer) => Promise<void>
): Promise<void> {
this.logger.info(`Initializing ${this.serviceConfig.serviceName} service...`);
try {
// Create and initialize container
this.logger.debug('Creating DI container...');
this.container = await containerFactory(this.config);
this.serviceContainer = this.container.resolve('serviceContainer');
this.logger.info('DI container created and initialized');
// Call container ready hook
if (this.hooks.onContainerReady) {
await this.hooks.onContainerReady(this.serviceContainer);
}
// Create Hono application
this.app = this.createApp();
// Call app ready hook
if (this.hooks.onAppReady) {
await this.hooks.onAppReady(this.app, this.serviceContainer);
}
// Initialize handlers if enabled
if (this.serviceConfig.enableHandlers && handlerInitializer) {
this.logger.debug('Initializing handlers...');
await handlerInitializer(this.serviceContainer);
this.logger.info('Handlers initialized');
}
// Create and mount routes
const routes = routeFactory(this.serviceContainer);
this.app.route('/', routes);
// Initialize scheduled jobs if enabled
if (this.serviceConfig.enableScheduledJobs) {
await this.initializeScheduledJobs();
}
// Call before start hook
if (this.hooks.onBeforeStart) {
await this.hooks.onBeforeStart(this.app, this.serviceContainer);
}
// Register shutdown handlers
this.registerShutdownHandlers();
// Start HTTP server
const port = this.config.service.port;
this.server = Bun.serve({
port,
fetch: this.app.fetch,
development: this.config.environment === 'development',
});
this.logger.info(`${this.serviceConfig.serviceName} service started on port ${port}`);
// Call started hook
if (this.hooks.onStarted) {
await this.hooks.onStarted(port);
}
} catch (error) {
console.error('DETAILED ERROR:', error);
this.logger.error('Failed to start service', {
error: error instanceof Error ? error.message : String(error),
stack: error instanceof Error ? error.stack : undefined,
details: JSON.stringify(error, null, 2),
});
throw error;
}
}
/**
* Initialize scheduled jobs from handler registry
*/
private async initializeScheduledJobs(): Promise<void> {
if (!this.container) {
throw new Error('Container not initialized');
}
this.logger.debug('Creating scheduled jobs from registered handlers...');
const { handlerRegistry } = await import('@stock-bot/types');
const allHandlers = handlerRegistry.getAllHandlersWithSchedule();
let totalScheduledJobs = 0;
for (const [handlerName, config] of allHandlers) {
if (config.scheduledJobs && config.scheduledJobs.length > 0) {
const queueManager = this.container.resolve('queueManager');
if (!queueManager) {
this.logger.error('Queue manager is not initialized, cannot create scheduled jobs');
continue;
}
const queue = queueManager.getQueue(handlerName);
for (const scheduledJob of config.scheduledJobs) {
// Include handler and operation info in job data
const jobData = {
handler: handlerName,
operation: scheduledJob.operation,
payload: scheduledJob.payload,
};
// Build job options from scheduled job config
const jobOptions = {
priority: scheduledJob.priority,
delay: scheduledJob.delay,
repeat: {
immediately: scheduledJob.immediately,
},
};
await queue.addScheduledJob(
scheduledJob.operation,
jobData,
scheduledJob.cronPattern,
jobOptions
);
totalScheduledJobs++;
this.logger.debug('Scheduled job created', {
handler: handlerName,
operation: scheduledJob.operation,
cronPattern: scheduledJob.cronPattern,
immediately: scheduledJob.immediately,
priority: scheduledJob.priority,
});
}
}
}
this.logger.info('Scheduled jobs created', { totalJobs: totalScheduledJobs });
// Start queue workers
this.logger.debug('Starting queue workers...');
const queueManager = this.container.resolve('queueManager');
if (queueManager) {
queueManager.startAllWorkers();
this.logger.info('Queue workers started');
}
}
/**
* Stop the service gracefully
*/
async stop(): Promise<void> {
this.logger.info(`Stopping ${this.serviceConfig.serviceName} service...`);
await this.shutdown.shutdown();
}
/**
* Get the service container (for testing or advanced use cases)
*/
getServiceContainer(): IServiceContainer | null {
return this.serviceContainer;
}
/**
* Get the Hono app (for testing or advanced use cases)
*/
getApp(): Hono | null {
return this.app;
}
/**
* ServiceApplication - Common service initialization and lifecycle management
* Encapsulates common patterns for Hono-based microservices
*/
import { Hono } from 'hono';
import { cors } from 'hono/cors';
import { getLogger, setLoggerConfig, shutdownLoggers, type Logger } from '@stock-bot/logger';
import { Shutdown } from '@stock-bot/shutdown';
import type { AppConfig as StockBotAppConfig, UnifiedAppConfig } from '@stock-bot/config';
import { toUnifiedConfig } from '@stock-bot/config';
import type { IServiceContainer } from '@stock-bot/handlers';
import type { ServiceContainer } from './awilix-container';
/**
* Configuration for ServiceApplication
*/
export interface ServiceApplicationConfig {
/** Service name for logging and identification */
serviceName: string;
/** CORS configuration - if not provided, uses permissive defaults */
corsConfig?: Parameters<typeof cors>[0];
/** Whether to enable handler initialization */
enableHandlers?: boolean;
/** Whether to enable scheduled job creation */
enableScheduledJobs?: boolean;
/** Custom shutdown timeout in milliseconds */
shutdownTimeout?: number;
/** Service metadata for info endpoint */
serviceMetadata?: {
version?: string;
description?: string;
endpoints?: Record<string, string>;
};
/** Whether to add a basic info endpoint at root */
addInfoEndpoint?: boolean;
}
/**
* Lifecycle hooks for service customization
*/
export interface ServiceLifecycleHooks {
/** Called after container is created but before routes */
onContainerReady?: (container: IServiceContainer) => Promise<void> | void;
/** Called after app is created but before routes are mounted */
onAppReady?: (app: Hono, container: IServiceContainer) => Promise<void> | void;
/** Called after routes are mounted but before server starts */
onBeforeStart?: (app: Hono, container: IServiceContainer) => Promise<void> | void;
/** Called after successful server startup */
onStarted?: (port: number) => Promise<void> | void;
/** Called during shutdown before cleanup */
onBeforeShutdown?: () => Promise<void> | void;
}
/**
* ServiceApplication - Manages the complete lifecycle of a microservice
*/
export class ServiceApplication {
private config: UnifiedAppConfig;
private serviceConfig: ServiceApplicationConfig;
private hooks: ServiceLifecycleHooks;
private logger: Logger;
private container: ServiceContainer | null = null;
private serviceContainer: IServiceContainer | null = null;
private app: Hono | null = null;
private server: ReturnType<typeof Bun.serve> | null = null;
private shutdown: Shutdown;
constructor(
config: StockBotAppConfig | UnifiedAppConfig,
serviceConfig: ServiceApplicationConfig,
hooks: ServiceLifecycleHooks = {}
) {
// Convert to unified config
this.config = toUnifiedConfig(config);
// Ensure service name is set in config
if (!this.config.service.serviceName) {
this.config.service.serviceName = serviceConfig.serviceName;
}
this.serviceConfig = {
shutdownTimeout: 15000,
enableHandlers: false,
enableScheduledJobs: false,
addInfoEndpoint: true,
...serviceConfig,
};
this.hooks = hooks;
// Initialize logger configuration
this.configureLogger();
this.logger = getLogger(this.serviceConfig.serviceName);
// Initialize shutdown manager
this.shutdown = Shutdown.getInstance({
timeout: this.serviceConfig.shutdownTimeout
});
}
/**
* Configure logger based on application config
*/
private configureLogger(): void {
if (this.config.log) {
setLoggerConfig({
logLevel: this.config.log.level,
logConsole: true,
logFile: false,
environment: this.config.environment,
hideObject: this.config.log.hideObject,
});
}
}
/**
* Create and configure Hono application with CORS
*/
private createApp(): Hono {
const app = new Hono();
// Add CORS middleware with service-specific or default configuration
const corsConfig = this.serviceConfig.corsConfig || {
origin: '*',
allowMethods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS', 'PATCH'],
allowHeaders: ['Content-Type', 'Authorization'],
credentials: false,
};
app.use('*', cors(corsConfig));
// Add basic info endpoint if enabled
if (this.serviceConfig.addInfoEndpoint) {
const metadata = this.serviceConfig.serviceMetadata || {};
app.get('/', c => {
return c.json({
name: this.serviceConfig.serviceName,
version: metadata.version || '1.0.0',
description: metadata.description,
status: 'running',
timestamp: new Date().toISOString(),
endpoints: metadata.endpoints || {},
});
});
}
return app;
}
/**
* Register graceful shutdown handlers
*/
private registerShutdownHandlers(): void {
// Priority 1: Queue system (highest priority)
if (this.serviceConfig.enableScheduledJobs) {
this.shutdown.onShutdownHigh(async () => {
this.logger.info('Shutting down queue system...');
try {
const queueManager = this.container?.resolve('queueManager');
if (queueManager) {
await queueManager.shutdown();
}
this.logger.info('Queue system shut down');
} catch (error) {
this.logger.error('Error shutting down queue system', { error });
}
}, 'Queue System');
}
// Priority 1: HTTP Server (high priority)
this.shutdown.onShutdownHigh(async () => {
if (this.server) {
this.logger.info('Stopping HTTP server...');
try {
this.server.stop();
this.logger.info('HTTP server stopped');
} catch (error) {
this.logger.error('Error stopping HTTP server', { error });
}
}
}, 'HTTP Server');
// Custom shutdown hook
if (this.hooks.onBeforeShutdown) {
this.shutdown.onShutdownHigh(async () => {
try {
await this.hooks.onBeforeShutdown!();
} catch (error) {
this.logger.error('Error in custom shutdown hook', { error });
}
}, 'Custom Shutdown');
}
// Priority 2: Services and connections (medium priority)
this.shutdown.onShutdownMedium(async () => {
this.logger.info('Disposing services and connections...');
try {
if (this.container) {
// Disconnect database clients
const mongoClient = this.container.resolve('mongoClient');
if (mongoClient?.disconnect) {
await mongoClient.disconnect();
}
const postgresClient = this.container.resolve('postgresClient');
if (postgresClient?.disconnect) {
await postgresClient.disconnect();
}
const questdbClient = this.container.resolve('questdbClient');
if (questdbClient?.disconnect) {
await questdbClient.disconnect();
}
this.logger.info('All services disposed successfully');
}
} catch (error) {
this.logger.error('Error disposing services', { error });
}
}, 'Services');
// Priority 3: Logger shutdown (lowest priority - runs last)
this.shutdown.onShutdownLow(async () => {
try {
this.logger.info('Shutting down loggers...');
await shutdownLoggers();
// Don't log after shutdown
} catch {
// Silently ignore logger shutdown errors
}
}, 'Loggers');
}
/**
* Start the service with full initialization
*/
async start(
containerFactory: (config: UnifiedAppConfig) => Promise<ServiceContainer>,
routeFactory: (container: IServiceContainer) => Hono,
handlerInitializer?: (container: IServiceContainer) => Promise<void>
): Promise<void> {
this.logger.info(`Initializing ${this.serviceConfig.serviceName} service...`);
try {
// Create and initialize container
this.logger.debug('Creating DI container...');
// Config already has service name from constructor
this.container = await containerFactory(this.config);
this.serviceContainer = this.container.resolve('serviceContainer');
this.logger.info('DI container created and initialized');
// Call container ready hook
if (this.hooks.onContainerReady) {
await this.hooks.onContainerReady(this.serviceContainer);
}
// Create Hono application
this.app = this.createApp();
// Call app ready hook
if (this.hooks.onAppReady) {
await this.hooks.onAppReady(this.app, this.serviceContainer);
}
// Initialize handlers if enabled
if (this.serviceConfig.enableHandlers && handlerInitializer) {
this.logger.debug('Initializing handlers...');
await handlerInitializer(this.serviceContainer);
this.logger.info('Handlers initialized');
}
// Create and mount routes
const routes = routeFactory(this.serviceContainer);
this.app.route('/', routes);
// Initialize scheduled jobs if enabled
if (this.serviceConfig.enableScheduledJobs) {
await this.initializeScheduledJobs();
}
// Call before start hook
if (this.hooks.onBeforeStart) {
await this.hooks.onBeforeStart(this.app, this.serviceContainer);
}
// Register shutdown handlers
this.registerShutdownHandlers();
// Start HTTP server
const port = this.config.service.port;
this.server = Bun.serve({
port,
fetch: this.app.fetch,
development: this.config.environment === 'development',
});
this.logger.info(`${this.serviceConfig.serviceName} service started on port ${port}`);
// Call started hook
if (this.hooks.onStarted) {
await this.hooks.onStarted(port);
}
} catch (error) {
console.error('DETAILED ERROR:', error);
this.logger.error('Failed to start service', {
error: error instanceof Error ? error.message : String(error),
stack: error instanceof Error ? error.stack : undefined,
details: JSON.stringify(error, null, 2),
});
throw error;
}
}
/**
* Initialize scheduled jobs from handler registry
*/
private async initializeScheduledJobs(): Promise<void> {
if (!this.container) {
throw new Error('Container not initialized');
}
this.logger.debug('Creating scheduled jobs from registered handlers...');
const { handlerRegistry } = await import('@stock-bot/types');
const allHandlers = handlerRegistry.getAllHandlersWithSchedule();
let totalScheduledJobs = 0;
for (const [handlerName, config] of allHandlers) {
if (config.scheduledJobs && config.scheduledJobs.length > 0) {
const queueManager = this.container.resolve('queueManager');
if (!queueManager) {
this.logger.error('Queue manager is not initialized, cannot create scheduled jobs');
continue;
}
const queue = queueManager.getQueue(handlerName);
for (const scheduledJob of config.scheduledJobs) {
// Include handler and operation info in job data
const jobData = {
handler: handlerName,
operation: scheduledJob.operation,
payload: scheduledJob.payload,
};
// Build job options from scheduled job config
const jobOptions = {
priority: scheduledJob.priority,
delay: scheduledJob.delay,
repeat: {
immediately: scheduledJob.immediately,
},
};
await queue.addScheduledJob(
scheduledJob.operation,
jobData,
scheduledJob.cronPattern,
jobOptions
);
totalScheduledJobs++;
this.logger.debug('Scheduled job created', {
handler: handlerName,
operation: scheduledJob.operation,
cronPattern: scheduledJob.cronPattern,
immediately: scheduledJob.immediately,
priority: scheduledJob.priority,
});
}
}
}
this.logger.info('Scheduled jobs created', { totalJobs: totalScheduledJobs });
// Start queue workers
this.logger.debug('Starting queue workers...');
const queueManager = this.container.resolve('queueManager');
if (queueManager) {
queueManager.startAllWorkers();
this.logger.info('Queue workers started');
}
}
/**
* Stop the service gracefully
*/
async stop(): Promise<void> {
this.logger.info(`Stopping ${this.serviceConfig.serviceName} service...`);
await this.shutdown.shutdown();
}
/**
* Get the service container (for testing or advanced use cases)
*/
getServiceContainer(): IServiceContainer | null {
return this.serviceContainer;
}
/**
* Get the Hono app (for testing or advanced use cases)
*/
getApp(): Hono | null {
return this.app;
}
}

20
libs/services/queue/src/service-registry.ts
View file

@ -45,7 +45,25 @@ export const SERVICE_REGISTRY: Record<string, ServiceConfig> = {
cachePrefix: 'cache:api',
producerOnly: true,
},
// Add more services as needed
// Add aliases for services with different naming conventions
'webApi': {
db: 3,
queuePrefix: 'bull:api',
cachePrefix: 'cache:api',
producerOnly: true,
},
'dataIngestion': {
db: 1,
queuePrefix: 'bull:di',
cachePrefix: 'cache:di',
handlers: ['ceo', 'qm', 'webshare', 'ib', 'proxy'],
},
'dataPipeline': {
db: 2,
queuePrefix: 'bull:dp',
cachePrefix: 'cache:dp',
handlers: ['exchanges', 'symbols'],
},
};
/**